diff --git a/docs/best-practices.md b/docs/best-practices.md
index 588594775..fff6f25d6 100644
--- a/docs/best-practices.md
+++ b/docs/best-practices.md
@@ -106,6 +106,29 @@ single response.
 This will help to reduce the response time and the load on the server, and improve the performance of your
 application.
 
+### Load relations for a whole result set, not per record
+
+Touching a relation on each record of a result set issues one query per record — the N+1 problem. Name
+the relation on the query instead, and it is resolved for the entire batch in a constant number of
+queries:
+
+```cpp
+auto albums = dm.Query<Album>().With<&Album::tracks>().All(); // 2 queries, not 1 + N
+```
+
+A nested relation needs the whole path named — `.With<&Track::album, &Album::artist>()` — because
+each record holds its own copy of the target, so one level of eager loading leaves the level below it
+loading per record. `DataMapperOptions { .eagerLoadDepth = N }` loads everything reachable instead,
+at the cost of fetching more than you asked for.
+
+See [Eager loading of relations](usage.md). Two things compound with it:
+
+- **Index your foreign keys.** `CreateTable<Record>()` emits an index for every `BelongsTo` column,
+  because no supported engine indexes a foreign key implicitly. Tables created by hand, or by an older
+  version of Lightweight, need that index added — without it every relation query is a full table scan.
+- **Prove the absence of N+1 in tests.** A `SqlLogger` subclass counting `OnPrepare`/`OnExecuteDirect`
+  turns "this endpoint issues two queries" into an assertion instead of an assumption.
+
 ### Let block-prefetch cut network round-trips
 
 Per-row fetch loops issue one `SQLFetch` (one network round-trip) per row. Lightweight transparently
diff --git a/docs/usage.md b/docs/usage.md
index 079b88870..27151ae1f 100644
--- a/docs/usage.md
+++ b/docs/usage.md
@@ -186,6 +186,87 @@ void BulkInsert(DataMapper& dm, std::vector<Person> const& people)
 > records (treat them as write-only inputs), and `UpdateAll` writes a uniform set of columns for every
 > row rather than only the per-record modified ones. The range must be contiguous.
 
+### Eager loading of relations (`With<>()`)
+
+Accessing a relation on a query result loads it on demand — one query per record. Over a result set of
+N records that is the N+1 problem: reading `album.tracks` for 1000 albums issues 1001 queries.
+`With<&Record::relation>()` instead resolves the relation for the whole result set once it has been
+materialized, using `WHERE <key> IN (...)`:
+
+```cpp
+// Two queries in total, whatever the number of albums: one for the albums, one for all their tracks.
+auto albums = dm.Query<Album>()
+                .With<&Album::tracks>()    // HasMany
+                .With<&Album::artist>()    // BelongsTo
+                .All();
+
+for (auto& album: albums)
+    for (auto const& track: album.tracks.All())  // already loaded, no query
+        std::println("{} - {}", album.title, track->title);
+```
+
+- Supported for `BelongsTo` and `HasMany`. `HasOneThrough`, `HasManyThrough` and `CompositeForeignKey`
+  still load on demand; naming one of them in `With<>()` is a compile error rather than a silent
+  fallback.
+- Naming several relations forms a **path**, which is what a nested relation needs — see below.
+- Calls chain, one per relation to load. It applies to `All()`, `First()`, `First(n)` and `Range()`.
+- The `IN` predicate is chunked (see `SqlQueryFormatter::MaxInPredicateValues`, 1000 by default), so a
+  large batch costs one query per chunk — a constant number of queries per relation, never one per
+  record.
+- A `BelongsTo` whose foreign key is `NULL`, and an owner with no children, are handled without an
+  extra query: the childless owner's relation is marked loaded-and-empty rather than left to query for
+  a result already known.
+- Relations that were *not* named keep their on-demand behaviour. Combining `With<>()` with
+  `DataMapperOptions { .loadRelations = false }` therefore turns any unrequested relation access into a
+  `SqlRequireLoadedError` instead of a silent query — useful to prove a code path issues no N+1.
+
+#### Nested relations
+
+Eager-loading one level is not enough for a chain. Every record holds its *own copy* of its
+`BelongsTo` target, so reaching a relation of that copy runs the copy's own lazy loader — the N+1
+simply moves one level down. Name the whole path instead:
+
+```cpp
+auto tracks = dm.Query<Track>()
+                .With<&Track::album>()                   // 1 query for all albums
+                .With<&Track::album, &Album::artist>()   // 1 query for all those albums' artists
+                .All();
+
+for (auto& track: tracks)
+    std::println("{} - {}", track.album.Record().title,
+                 track.album.Record().artist.Record().name);   // no queries here
+```
+
+Three queries in total, for any number of tracks. Each level is resolved for every record reached by
+the level above it, at once. A path may also run through the "many" side
+(`.With<&Album::tracks, &Track::genre>()`): the middle level fans out, and the level below it is
+still one query rather than one per child.
+
+Already-loaded relations are skipped, so overlapping paths (`.With<&A::b>()` next to
+`.With<&A::b, &B::c>()`) do not fetch `b` twice.
+
+#### Loading everything reachable
+
+When a whole object graph is wanted rather than named paths, set a depth on the query instead:
+
+```cpp
+// Tracks, their albums and categories, and those albums' artists - a constant number of queries.
+auto tracks = dm.Query<Track, DataMapperOptions { .eagerLoadDepth = 2 }>().All();
+```
+
+`eagerLoadDepth` batch-loads *every* `BelongsTo` and `HasMany` reachable within that many levels.
+Prefer `With<>()` when only part of the graph is needed: the depth walk fetches more rows, and
+instantiates the loader for the whole reachable relation graph, which costs compile time. The depth
+is what bounds both — and what lets a cyclic graph (a self-referencing record, or `A → B → A`)
+terminate, since the recursion is cut at a compile-time constant.
+
+Measured on 1000 owners with 10 children each, comparing the on-demand path with `With<>()`:
+
+| relation | queries before | queries after | SQLite3 | PostgreSQL | MS SQL Server |
+|---|---:|---:|---:|---:|---:|
+| `HasMany` | 1001 | 2 | 8.7x | 45x | 45x |
+| `BelongsTo` | 10001 | 2 | 37x | 464x | 407x |
+
 ## Simple row retrieval via structs
 
 When only read access is needed, you can use a simple `struct` to represent the row,
diff --git a/src/Lightweight/DataMapper/BelongsTo.hpp b/src/Lightweight/DataMapper/BelongsTo.hpp
index 333efd834..11fcb22b4 100644
--- a/src/Lightweight/DataMapper/BelongsTo.hpp
+++ b/src/Lightweight/DataMapper/BelongsTo.hpp
@@ -316,6 +316,24 @@ class BelongsTo
         return static_cast<bool>(_referencedFieldValue);
     }
 
+    /// @brief Returns the already-loaded referenced record, or `nullptr` when none is loaded.
+    ///
+    /// Unlike `Record()`, this never runs the on-demand loader: it reports what is present right
+    /// now. That is what lets the batched relation loading walk one level deeper (`With<A, B>()`)
+    /// without turning the walk itself into the N+1 it exists to remove.
+    ///
+    /// @return Pointer to the loaded record, or `nullptr` if the relation is unloaded or NULL.
+    [[nodiscard]] LIGHTWEIGHT_FORCE_INLINE constexpr ReferencedRecord* LoadedRecord() noexcept
+    {
+        return _record.get();
+    }
+
+    /// @copydoc LoadedRecord()
+    [[nodiscard]] LIGHTWEIGHT_FORCE_INLINE constexpr ReferencedRecord const* LoadedRecord() const noexcept
+    {
+        return _record.get();
+    }
+
     /// Emplaces a record into the relationship. This will mark the relationship as loaded.
     [[nodiscard]] LIGHTWEIGHT_FORCE_INLINE constexpr ReferencedRecord& EmplaceRecord()
     {
diff --git a/src/Lightweight/DataMapper/DataMapper.hpp b/src/Lightweight/DataMapper/DataMapper.hpp
index 145f38e33..113e333c2 100644
--- a/src/Lightweight/DataMapper/DataMapper.hpp
+++ b/src/Lightweight/DataMapper/DataMapper.hpp
@@ -38,6 +38,13 @@ namespace Lightweight
 
 namespace detail
 {
+    /// The first element of a non-empty pack of relation members, i.e. the head of a relation path.
+    ///
+    /// Declared up here because `With<>()` names it, and a non-dependent name inside a template is
+    /// looked up where the template is defined, not where it is instantiated.
+    template <auto First, auto...>
+    inline constexpr auto FirstOf = First;
+
     // Converts a container of T to a container of std::shared_ptr<T>.
     template <template <typename> class Allocator, template <typename, typename> class Container, typename Object>
     auto ToSharedPtrList(Container<Object, Allocator<Object>> container)
@@ -583,6 +590,11 @@ class DataMapper
     [[nodiscard]] Async::Task<void> LoadRelationsAsync(Record& record);
 
   private:
+    // The fluent query builders drive the batched relation preloading (`With<>()`), which is an
+    // internal entry point rather than part of the mapper's public surface.
+    template <typename BuilderRecord, typename Derived, DataMapperOptions BuilderQueryOptions>
+    friend class SqlCoreDataMapperQueryBuilder;
+
     /// Builds the comma-separated, fully-qualified (`"Table"."Column"`) field list for @p Record.
     ///
     /// Shared by @c Query and @c QueryAsync so the SELECT projection is produced in exactly one place.
@@ -659,6 +671,74 @@ class DataMapper
     template <typename Record, typename OtherRecord, auto InverseSelector>
     void LoadHasMany(Record& record, HasMany<OtherRecord, InverseSelector>& field);
 
+    /// @brief Eagerly resolves one relation for a whole batch of records, in a bounded number of queries.
+    ///
+    /// This is the engine behind `Query<Record>().With<&Record::relation>()`. Where the lazy loaders
+    /// installed by `ConfigureRelationAutoLoading()` issue one query per record touched - the N+1 - this
+    /// issues one query per chunk of the batch (see `SqlQueryFormatter::MaxInPredicateValues()`) and
+    /// distributes the resulting rows into the records in memory.
+    ///
+    /// The batch is addressed by pointer rather than as a contiguous range: past the first level of
+    /// a relation path the targets are not contiguous - every owner holds its own copy of a
+    /// `BelongsTo` target, and a `HasMany` list holds `shared_ptr`s - so only their addresses can be
+    /// gathered. `SqlCoreDataMapperQueryBuilder::RunRelationPreloaders()` adapts a freshly
+    /// materialized result set onto this.
+    ///
+    /// @tparam Record     The record type owning the relation.
+    /// @tparam FieldIndex Member index, within @p Record, of the relation to load.
+    /// @param records     The batch to resolve the relation for.
+    template <typename Record, size_t FieldIndex>
+    void PreloadRelation(std::span<Record* const> records);
+
+    /// Batch counterpart of `LoadBelongsTo`: one query per chunk of distinct foreign keys.
+    /// @see PreloadRelation
+    template <typename Record, size_t FieldIndex>
+    void PreloadBelongsTo(std::span<Record* const> records);
+
+    /// Batch counterpart of `LoadHasMany`: one query per chunk of owner primary keys, then the rows
+    /// are grouped by the inverse foreign key and handed to each owner.
+    /// @see PreloadRelation
+    template <typename Record, size_t FieldIndex>
+    void PreloadHasMany(std::span<Record* const> records);
+
+    /// @brief Eagerly resolves a *path* of relations, one bounded set of queries per level.
+    ///
+    /// Backs `Query<Record>().With<&Track::album, &Album::artist>()`: the first relation is resolved
+    /// for the whole batch, the loaded targets are then gathered and the next relation resolved for
+    /// all of them at once, and so on. Every level costs a constant number of queries, so a two-level
+    /// path over any number of records is three queries in total.
+    ///
+    /// @tparam Record       The record type owning the first relation of the path.
+    /// @tparam FieldIndex   Member index, within @p Record, of the first relation.
+    /// @tparam RestOfPath   The remaining relations, each a member of the preceding target type.
+    /// @param records       The batch to resolve the path for.
+    template <typename Record, size_t FieldIndex, auto... RestOfPath>
+    void PreloadRelationPath(std::span<Record* const> records);
+
+    /// @brief Eagerly resolves *every* supported relation reachable within @p Depth levels.
+    ///
+    /// Backs `DataMapperOptions { .eagerLoadDepth = N }`. Each level costs a bounded number of
+    /// queries per relation, never one per record. `HasOneThrough`, `HasManyThrough` and
+    /// `CompositeForeignKey` are skipped and keep loading on demand.
+    ///
+    /// @tparam Record The record type whose relations are resolved.
+    /// @tparam Depth  Remaining levels to descend; `0` stops the recursion, which is what keeps a
+    ///                cyclic relation graph (a self-referencing record, or A -> B -> A) from
+    ///                instantiating forever.
+    /// @param records The batch to resolve the relations for.
+    template <typename Record, size_t Depth>
+    void PreloadAllRelations(std::span<Record* const> records);
+
+    /// Appends the addresses of the records loaded into relation @p FieldIndex of every record in
+    /// @p records, so the next level of a path can be resolved for all of them at once.
+    ///
+    /// @tparam Record     The record type owning the relation.
+    /// @tparam FieldIndex Member index, within @p Record, of the relation to walk into.
+    /// @param records     The batch to collect from.
+    /// @return The loaded targets; records whose relation is unloaded or NULL contribute nothing.
+    template <typename Record, size_t FieldIndex>
+    [[nodiscard]] static auto CollectRelationTargets(std::span<Record* const> records);
+
     template <typename ReferencedRecord, typename ThroughSpec, typename Record, auto OwnerSelector, auto ThroughSelector>
     void LoadHasOneThrough(Record& record,
                            HasOneThrough<ReferencedRecord, ThroughSpec, OwnerSelector, ThroughSelector>& field);
@@ -673,6 +753,15 @@ class DataMapper
     template <typename OwnerRecord, typename OtherRecord, auto InverseSelector>
     SqlSelectQueryBuilder BuildHasManySelectQuery();
 
+    /// @brief Builds a `SELECT <all storage columns> FROM <table of @p Record>`, without any predicate.
+    ///
+    /// The starting point for the batched relation queries, which add their own `WHERE ... IN (...)`.
+    ///
+    /// @tparam Record The record type whose table and columns are selected.
+    /// @return The select query builder, ready for further clauses.
+    template <typename Record>
+    SqlSelectQueryBuilder BuildRecordSelectQuery();
+
     template <typename ReferencedRecord, typename ThroughRecord, typename Record, auto OwnerSelector, auto ThroughSelector>
     SqlSelectQueryBuilder BuildHasOneThroughSelectQuery();
 
@@ -1144,6 +1233,59 @@ size_t SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>::CountImpl()
     return 0;
 }
 
+template <typename Record, typename Derived, DataMapperOptions QueryOptions>
+template <auto... RelationPath>
+    requires DataMapperRecord<Record> && (sizeof...(RelationPath) >= 1)
+Derived& SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>::With()
+{
+    constexpr auto FirstField = detail::FirstOf<RelationPath...>;
+
+#if defined(LIGHTWEIGHT_CXX26_REFLECTION)
+    static_assert(std::same_as<typename[:std::meta::parent_of(FirstField):], Record>,
+                  "The first relation named by With<>() must be a member of the record being queried");
+#else
+    // remove_cv_t: FirstField is a constexpr variable, so decltype() yields a *const* member pointer
+    // type, which MemberClassType is not specialized for.
+    static_assert(std::same_as<MemberClassType<std::remove_cv_t<decltype(FirstField)>>, Record>,
+                  "The first relation named by With<>() must be a member of the record being queried");
+#endif
+
+    // A capture-less lambda, so this decays to a plain function pointer: the whole path is known at
+    // compile time, so nothing has to be carried at run time beyond which instantiation to call.
+    _relationPreloaders.emplace_back(+[](DataMapper& dm, std::span<Record* const> records) {
+        [&]<auto Head, auto... Tail>() {
+            dm.template PreloadRelationPath<Record, MemberIndexOf<Head>, Tail...>(records);
+        }.template operator()<RelationPath...>();
+    });
+
+    return static_cast<Derived&>(*this);
+}
+
+template <typename Record, typename Derived, DataMapperOptions QueryOptions>
+void SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>::RunRelationPreloaders(std::span<Record> records)
+{
+    if constexpr (QueryOptions.eagerLoadDepth == 0)
+        if (_relationPreloaders.empty())
+            return;
+
+    if (records.empty())
+        return;
+
+    auto pointers = std::vector<Record*> {};
+    pointers.reserve(records.size());
+    for (auto& record: records)
+        pointers.emplace_back(&record);
+    auto const batch = std::span<Record* const> { pointers };
+
+    // The named paths run first: the batched loaders skip a relation that is already loaded, so a
+    // relation this query asked for by name is not fetched a second time by the depth walk below.
+    for (auto const preloader: _relationPreloaders)
+        preloader(_dm, batch);
+
+    if constexpr (QueryOptions.eagerLoadDepth > 0)
+        _dm.template PreloadAllRelations<Record, QueryOptions.eagerLoadDepth>(batch);
+}
+
 template <typename Record, typename Derived, DataMapperOptions QueryOptions>
 bool SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>::ExistImpl()
 {
@@ -1206,6 +1348,7 @@ std::vector<Record> SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>
                 _dm.ConfigureRelationAutoLoading(record);
             }
         }
+        RunRelationPreloaders(records);
     }
     return records;
 }
@@ -1315,6 +1458,9 @@ std::optional<Record> SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOption
         if (record)
             _dm.ConfigureRelationAutoLoading(record.value());
     }
+    if constexpr (DataMapperRecord<Record>)
+        if (record)
+            RunRelationPreloaders(std::span<Record> { &record.value(), 1 });
     return record;
 }
 
@@ -1417,6 +1563,8 @@ std::vector<Record> SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>
         for (auto& record: records)
             _dm.ConfigureRelationAutoLoading(record);
     }
+    if constexpr (DataMapperRecord<Record>)
+        RunRelationPreloaders(records);
     return records;
 }
 
@@ -1445,6 +1593,8 @@ std::vector<Record> SqlCoreDataMapperQueryBuilder<Record, Derived, QueryOptions>
         for (auto& record: records)
             _dm.ConfigureRelationAutoLoading(record);
     }
+    if constexpr (DataMapperRecord<Record>)
+        RunRelationPreloaders(records);
     return records;
 }
 
@@ -2647,19 +2797,7 @@ void DataMapper::CallOnHasMany(Record& record, Callable const& callback)
     using ReferencedRecord = FieldType::ReferencedRecord;
 
     CallOnPrimaryKey(record, [&]<size_t PrimaryKeyIndex, typename PrimaryKeyType>(PrimaryKeyType const& primaryKeyField) {
-        auto query = _connection.Query(RecordTableName<ReferencedRecord>)
-                         .Select()
-                         .Build([&](auto& query) {
-                             EnumerateRecordMembers<ReferencedRecord>(
-                                 [&]<size_t ReferencedFieldIndex, typename ReferencedFieldType>() {
-                                     if constexpr (FieldWithStorage<ReferencedFieldType>)
-                                     {
-                                         query.Field(FieldNameAt<ReferencedFieldIndex, ReferencedRecord>);
-                                     }
-                                 });
-                         })
-                         .Where(InverseBelongsToFieldNameOf<Record, ReferencedRecord, InverseSelector>, SqlWildcard)
-                         .OrderBy(FieldNameAt<RecordPrimaryKeyIndex<ReferencedRecord>, ReferencedRecord>);
+        auto query = BuildHasManySelectQuery<Record, ReferencedRecord, InverseSelector>();
         callback(query, primaryKeyField);
     });
 }
@@ -2667,14 +2805,7 @@ void DataMapper::CallOnHasMany(Record& record, Callable const& callback)
 template <typename OwnerRecord, typename OtherRecord, auto InverseSelector>
 SqlSelectQueryBuilder DataMapper::BuildHasManySelectQuery()
 {
-    return _connection.Query(RecordTableName<OtherRecord>)
-        .Select()
-        .Build([](auto& q) {
-            EnumerateRecordMembers<OtherRecord>([&]<size_t I, typename F>() {
-                if constexpr (FieldWithStorage<F>)
-                    q.Field(FieldNameAt<I, OtherRecord>);
-            });
-        })
+    return BuildRecordSelectQuery<OtherRecord>()
         .Where(InverseBelongsToFieldNameOf<OwnerRecord, OtherRecord, InverseSelector>, SqlWildcard)
         .OrderBy(FieldNameAt<RecordPrimaryKeyIndex<OtherRecord>, OtherRecord>);
 }
@@ -2694,6 +2825,333 @@ void DataMapper::LoadHasMany(Record& record, HasMany<OtherRecord, InverseSelecto
         });
 }
 
+namespace detail
+{
+    /// @brief Invokes @p callable once per chunk of at most @p chunkSize consecutive elements of @p values.
+    ///
+    /// Used to keep a generated `WHERE ... IN (...)` predicate inside what the dialect accepts (see
+    /// @ref SqlQueryFormatter::MaxInPredicateValues) while still resolving a whole batch in a number of
+    /// queries that depends on the chunk size, not on the number of records.
+    ///
+    /// @param values The values to split.
+    /// @param chunkSize Maximum number of values per chunk; zero is treated as "all in one chunk".
+    /// @param callable Invoked with a `std::span` over each chunk, in order.
+    template <typename T, typename Callable>
+    void ForEachChunk(std::span<T const> values, size_t chunkSize, Callable const& callable)
+    {
+        auto const stride = chunkSize == 0 ? values.size() : chunkSize;
+        for (size_t offset = 0; offset < values.size(); offset += stride)
+            callable(values.subspan(offset, (std::min) (stride, values.size() - offset)));
+    }
+} // namespace detail
+
+template <typename Record>
+SqlSelectQueryBuilder DataMapper::BuildRecordSelectQuery()
+{
+    return _connection.Query(RecordTableName<Record>).Select().Build([](auto& query) {
+        EnumerateRecordMembers<Record>([&query]<size_t I, typename FieldType>() {
+            if constexpr (FieldWithStorage<FieldType>)
+                query.Field(FieldNameAt<I, Record>);
+        });
+    });
+}
+
+template <typename Record, size_t FieldIndex>
+void DataMapper::PreloadRelation(std::span<Record* const> records)
+{
+    using FieldType = RecordMemberTypeOf<FieldIndex, Record>;
+
+    static_assert(IsBelongsTo<FieldType> || IsHasMany<FieldType>,
+                  "Eager loading via With<>() is supported for BelongsTo and HasMany relations. "
+                  "HasOneThrough, HasManyThrough and CompositeForeignKey still load on demand.");
+
+    // Defensive rather than reachable: every caller already returns on an empty batch
+    // (RunRelationPreloaders, PreloadAllRelations, and the `targets.empty()` guards in
+    // PreloadRelationPath / PreloadAllRelations). Kept so a future caller cannot trip over it, which
+    // is also why coverage reports never mark this line.
+    if (records.empty())
+        return;
+
+    if constexpr (IsBelongsTo<FieldType>)
+        PreloadBelongsTo<Record, FieldIndex>(records);
+    else if constexpr (IsHasMany<FieldType>)
+        PreloadHasMany<Record, FieldIndex>(records);
+}
+
+template <typename Record, size_t FieldIndex>
+auto DataMapper::CollectRelationTargets(std::span<Record* const> records)
+{
+    using FieldType = RecordMemberTypeOf<FieldIndex, Record>;
+    using TargetRecord = typename FieldType::ReferencedRecord;
+
+    auto targets = std::vector<TargetRecord*> {};
+    targets.reserve(records.size());
+
+    for (auto* record: records)
+    {
+        auto& field = GetRecordMemberAt<FieldIndex>(*record);
+        if constexpr (IsBelongsTo<FieldType>)
+        {
+            // LoadedRecord(), not Record(): asking through the accessor would run the on-demand
+            // loader for exactly the rows the batch failed to resolve, one query at a time.
+            if (auto* target = field.LoadedRecord(); target != nullptr)
+                targets.emplace_back(target);
+        }
+        else if constexpr (IsHasMany<FieldType>)
+        {
+            if (auto* loaded = field.LoadedRecords(); loaded != nullptr)
+                for (auto& element: *loaded)
+                    targets.emplace_back(element.get());
+        }
+    }
+
+    return targets;
+}
+
+template <typename Record, size_t FieldIndex, auto... RestOfPath>
+void DataMapper::PreloadRelationPath(std::span<Record* const> records)
+{
+    PreloadRelation<Record, FieldIndex>(records);
+
+    if constexpr (sizeof...(RestOfPath) > 0)
+    {
+        using FieldType = RecordMemberTypeOf<FieldIndex, Record>;
+        using TargetRecord = typename FieldType::ReferencedRecord;
+
+        auto targets = CollectRelationTargets<Record, FieldIndex>(records);
+        if (targets.empty())
+            return;
+
+        // Every owner holds its own copy of a BelongsTo target, so the same underlying row can
+        // appear many times here. That is not wasted work: the level below deduplicates the keys
+        // before querying, so the row is still fetched once and then handed to each copy.
+        // The static_assert below is what keeps the path honest: without it, a path element naming a
+        // member of some *other* record still compiles, because MemberIndexOf<> yields that member's
+        // index within its own class, which is then read as an index into TargetRecord - silently
+        // eager-loading whatever relation happens to sit at the same position (say
+        // `With<&Track::album, &Track::genre>()` resolving Album's member 2).
+        [&]<auto NextField, auto... Rest>() {
+#if defined(LIGHTWEIGHT_CXX26_REFLECTION)
+            static_assert(std::same_as<typename[:std::meta::parent_of(NextField):], TargetRecord>,
+                          "Each relation named after the first in With<>() must be a member of the "
+                          "preceding relation's referenced record");
+#else
+            static_assert(std::same_as<MemberClassType<std::remove_cv_t<decltype(NextField)>>, TargetRecord>,
+                          "Each relation named after the first in With<>() must be a member of the "
+                          "preceding relation's referenced record");
+#endif
+            PreloadRelationPath<TargetRecord, MemberIndexOf<NextField>, Rest...>(std::span<TargetRecord* const> { targets });
+        }.template operator()<RestOfPath...>();
+    }
+}
+
+template <typename Record, size_t Depth>
+void DataMapper::PreloadAllRelations(std::span<Record* const> records)
+{
+    if constexpr (Depth == 0)
+        return;
+    else
+    {
+        // As in PreloadRelation: both callers (RunRelationPreloaders and the recursive descent below)
+        // already guard against an empty batch, so this is belt-and-braces and stays uncovered.
+        if (records.empty())
+            return;
+
+        Reflection::template_for<0, RecordMemberCount<Record>>([&]<auto I>() {
+            using FieldType = RecordMemberTypeOf<I, Record>;
+            // Relations the batched loaders do not cover are left to their on-demand loaders rather
+            // than failing the whole query: this option means "load what can be loaded in bulk".
+            if constexpr (IsBelongsTo<FieldType> || IsHasMany<FieldType>)
+            {
+                PreloadRelation<Record, I>(records);
+
+                if constexpr (Depth > 1)
+                {
+                    using TargetRecord = typename FieldType::ReferencedRecord;
+                    auto targets = CollectRelationTargets<Record, I>(records);
+                    if (!targets.empty())
+                        PreloadAllRelations<TargetRecord, Depth - 1>(std::span<TargetRecord* const> { targets });
+                }
+            }
+        });
+    }
+}
+
+template <typename Record, size_t FieldIndex>
+void DataMapper::PreloadBelongsTo(std::span<Record* const> records)
+{
+    using FieldType = RecordMemberTypeOf<FieldIndex, Record>;
+    using ReferencedRecord = typename FieldType::ReferencedRecord;
+    using KeyType = RecordPrimaryKeyType<ReferencedRecord>;
+
+    static_assert(HasPrimaryKey<ReferencedRecord>,
+                  "Eager loading a BelongsTo requires the referenced record to have a primary key");
+
+    ZoneScopedN("DataMapper::PreloadBelongsTo");
+    ZoneTextObject(RecordTableName<ReferencedRecord>);
+
+    auto const primaryKeyOf = [](ReferencedRecord const& record) {
+        return GetPrimaryKeyField(record);
+    };
+
+    // The distinct, non-NULL foreign keys of the batch. Sorted and deduplicated rather than hashed:
+    // ordering is all a key column type has to provide, while std::hash is not specialized for every
+    // one of them. The deduplication is what makes this cheaper than the row count - many rows
+    // commonly point at the same parent, which is then fetched once instead of once per row.
+    auto keys = std::vector<KeyType> {};
+    keys.reserve(records.size());
+    for (auto const* record: records)
+    {
+        auto const& field = GetRecordMemberAt<FieldIndex>(*record);
+        // Already resolved - by a named With<>() path, or by an outer level of the depth walk. Asking
+        // again would be a second query for a row that is already in memory.
+        if (field.LoadedRecord() != nullptr)
+            continue;
+
+        auto const& value = field.Value();
+        if constexpr (FieldType::IsOptional)
+        {
+            if (value.has_value())
+                keys.emplace_back(*value);
+        }
+        else
+            keys.emplace_back(value);
+    }
+    std::ranges::sort(keys);
+    keys.erase(std::ranges::unique(keys).begin(), keys.end());
+    if (keys.empty())
+        return;
+
+    auto loaded = std::vector<ReferencedRecord> {};
+    loaded.reserve(keys.size());
+    detail::ForEachChunk(std::span<KeyType const> { keys },
+                         _connection.QueryFormatter().MaxInPredicateValues(),
+                         [&](std::span<KeyType const> chunk) {
+                             auto selectQuery = BuildRecordSelectQuery<ReferencedRecord>().WhereIn(
+                                 FieldNameAt<RecordPrimaryKeyIndex<ReferencedRecord>, ReferencedRecord>, chunk);
+                             auto rows = Query<ReferencedRecord>(selectQuery.All());
+                             loaded.insert(
+                                 loaded.end(), std::make_move_iterator(rows.begin()), std::make_move_iterator(rows.end()));
+                         });
+
+    std::ranges::sort(loaded, {}, primaryKeyOf);
+
+    for (auto* record: records)
+    {
+        auto& field = GetRecordMemberAt<FieldIndex>(*record);
+        // Only reachable from a batch mixing already-loaded and unloaded records: the key-collection
+        // loop above skips the loaded ones, and a batch in which *every* record is loaded has already
+        // returned at the `keys.empty()` check. No current call path builds such a mix, so this arm is
+        // not covered by the suite.
+        if (field.LoadedRecord() != nullptr)
+            continue;
+
+        auto const& value = field.Value();
+        if constexpr (FieldType::IsOptional)
+            if (!value.has_value())
+                continue;
+
+        auto const key = [&] {
+            if constexpr (FieldType::IsOptional)
+                return *value;
+            else
+                return value;
+        }();
+
+        auto const it = std::ranges::lower_bound(loaded, key, {}, primaryKeyOf);
+        // A foreign key pointing at a row that is not there is left unloaded rather than adopted as
+        // empty: a missing target row is a data-integrity problem for the accessor to surface, which
+        // is also what the on-demand path does.
+        if (it != loaded.end() && primaryKeyOf(*it) == key)
+            field.AdoptFetchedRecord(*it);
+    }
+}
+
+template <typename Record, size_t FieldIndex>
+void DataMapper::PreloadHasMany(std::span<Record* const> records)
+{
+    using FieldType = RecordMemberTypeOf<FieldIndex, Record>;
+    using ReferencedRecord = typename FieldType::ReferencedRecord;
+    using ReferencedRecordList = typename FieldType::ReferencedRecordList;
+    using KeyType = RecordPrimaryKeyType<Record>;
+
+    static_assert(HasPrimaryKey<Record>, "Eager loading a HasMany requires the owning record to have a primary key");
+
+    ZoneScopedN("DataMapper::PreloadHasMany");
+    ZoneTextObject(RecordTableName<ReferencedRecord>);
+
+    // Every fetched child carries its owner's key in the BelongsTo that backs this relation; that is
+    // what assigns the row to an owner below, without a second query to find out.
+    constexpr size_t InverseIndex = InverseBelongsToIndexOf<Record, ReferencedRecord, FieldType::InverseSelector>;
+
+    auto keys = std::vector<KeyType> {};
+    keys.reserve(records.size());
+    for (auto* record: records)
+        // See PreloadBelongsTo: a list already in memory is not fetched again.
+        if (GetRecordMemberAt<FieldIndex>(*record).LoadedRecords() == nullptr)
+            keys.emplace_back(GetPrimaryKeyField(*record));
+    std::ranges::sort(keys);
+    keys.erase(std::ranges::unique(keys).begin(), keys.end());
+    if (keys.empty())
+        return;
+
+    // One bucket per distinct owner key, in the same sorted order, so a child row finds its owner by
+    // binary search rather than by scanning the batch (which would be quadratic in the row count).
+    auto buckets = std::vector<ReferencedRecordList> {};
+    buckets.resize(keys.size());
+
+    detail::ForEachChunk(
+        std::span<KeyType const> { keys },
+        _connection.QueryFormatter().MaxInPredicateValues(),
+        [&](std::span<KeyType const> chunk) {
+            auto selectQuery =
+                BuildRecordSelectQuery<ReferencedRecord>()
+                    .WhereIn(InverseBelongsToFieldNameOf<Record, ReferencedRecord, FieldType::InverseSelector>, chunk)
+                    .OrderBy(FieldNameAt<RecordPrimaryKeyIndex<ReferencedRecord>, ReferencedRecord>);
+            for (auto& child: Query<ReferencedRecord>(selectQuery.All()))
+            {
+                auto const& ownerValue = GetRecordMemberAt<InverseIndex>(child).Value();
+                using InverseFieldType = RecordMemberTypeOf<InverseIndex, ReferencedRecord>;
+                if constexpr (InverseFieldType::IsOptional)
+                    if (!ownerValue.has_value())
+                        continue;
+
+                auto const ownerKey = [&] {
+                    if constexpr (InverseFieldType::IsOptional)
+                        return *ownerValue;
+                    else
+                        return ownerValue;
+                }();
+
+                auto const it = std::ranges::lower_bound(keys, ownerKey);
+                if (it != keys.end() && *it == ownerKey)
+                    buckets[static_cast<size_t>(std::distance(keys.begin(), it))].emplace_back(
+                        std::make_shared<ReferencedRecord>(std::move(child)));
+            }
+        });
+
+    for (auto* record: records)
+    {
+        auto& field = GetRecordMemberAt<FieldIndex>(*record);
+        // See PreloadBelongsTo: reachable only from a batch mixing loaded and unloaded records, which
+        // no current call path produces - a wholly-loaded batch returns at the `keys.empty()` check.
+        if (field.LoadedRecords() != nullptr)
+            continue;
+
+        // Likewise unreachable today: every record that got past the check above contributed its key
+        // to `keys` in the collection loop, so the lookup cannot miss. Kept as a guard for a future
+        // caller that hands in a batch whose keys were filtered elsewhere.
+        auto const it = std::ranges::lower_bound(keys, GetPrimaryKeyField(*record));
+        if (it == keys.end() || !(*it == GetPrimaryKeyField(*record)))
+            continue;
+
+        // Owners with no children are emplaced empty on purpose: leaving the relation unloaded would
+        // send the very first access back to the database for a result already known to be empty.
+        auto& bucket = buckets[static_cast<size_t>(std::distance(keys.begin(), it))];
+        field.Emplace(ReferencedRecordList { bucket });
+    }
+}
+
 template <typename ReferencedRecord, typename ThroughRecord, typename Record, auto OwnerSelector, auto ThroughSelector>
 SqlSelectQueryBuilder DataMapper::BuildHasOneThroughSelectQuery()
 {
diff --git a/src/Lightweight/DataMapper/HasMany.hpp b/src/Lightweight/DataMapper/HasMany.hpp
index ce1cb116b..f3dfb92cd 100644
--- a/src/Lightweight/DataMapper/HasMany.hpp
+++ b/src/Lightweight/DataMapper/HasMany.hpp
@@ -110,6 +110,17 @@ class HasMany
     /// Emplaces the given list of records.
     ReferencedRecordList& Emplace(ReferencedRecordList&& records) noexcept;
 
+    /// @brief Returns the already-loaded records, or `nullptr` when the relation is not loaded.
+    ///
+    /// Unlike `All()`, this never runs the on-demand loader: it reports what is present right now,
+    /// which is what lets the batched relation loading walk one level deeper (`With<A, B>()`).
+    ///
+    /// @return Pointer to the loaded list, or `nullptr` if the relation was never loaded.
+    [[nodiscard]] ReferencedRecordList* LoadedRecords() noexcept
+    {
+        return _records ? &*_records : nullptr;
+    }
+
     /// Retrieves the number of records in this 1-to-many relationship.
     [[nodiscard]] std::size_t Count() const noexcept;
 
diff --git a/src/Lightweight/DataMapper/QueryBuilders.hpp b/src/Lightweight/DataMapper/QueryBuilders.hpp
index 4402ee242..7ce5985be 100644
--- a/src/Lightweight/DataMapper/QueryBuilders.hpp
+++ b/src/Lightweight/DataMapper/QueryBuilders.hpp
@@ -11,6 +11,8 @@
 #include "Record.hpp"
 
 #include <cstdint>
+#include <span>
+#include <vector>
 
 namespace Lightweight
 {
@@ -24,6 +26,27 @@ struct DataMapperOptions
 {
     /// Whether to automatically load relations when querying records.
     bool loadRelations { true };
+
+    /// @brief How many levels of relations to eagerly batch-load after a query materializes.
+    ///
+    /// `0` (the default) leaves every relation to load on demand, one query per record touched.
+    /// A value of `N` resolves every `BelongsTo` and `HasMany` reachable within `N` levels for the
+    /// whole result set at once, at a bounded number of queries per relation per level:
+    ///
+    /// @code
+    /// // Tracks, their albums, and those albums' artists - a constant number of queries.
+    /// auto tracks = dm.Query<Track, DataMapperOptions { .eagerLoadDepth = 2 }>().All();
+    /// @endcode
+    ///
+    /// Use `With<>()` instead when only some relations are needed: this option descends into
+    /// *every* relation of every record it reaches, which is more queries and more rows than a
+    /// named path, and instantiates the loader for the whole reachable relation graph. The depth is
+    /// what bounds that: a cyclic graph (a self-referencing record, or A -> B -> A) terminates
+    /// because the recursion is cut at a compile-time constant.
+    ///
+    /// `HasOneThrough`, `HasManyThrough` and `CompositeForeignKey` are not batch-loadable and keep
+    /// their on-demand behaviour.
+    size_t eagerLoadDepth { 0 };
 };
 
 /// Selects whether a query builder's finisher methods execute synchronously or asynchronously.
@@ -52,6 +75,15 @@ class [[nodiscard]] SqlCoreDataMapperQueryBuilder: public SqlBasicSelectQueryBui
     std::string _fields;
     std::vector<SqlVariant> _boundInputs;
 
+    /// One batched relation loader, as requested by `With<&Record::relation>()`.
+    ///
+    /// A plain function pointer rather than a `std::function`: every `With<>()` names its relation at
+    /// compile time, so the loader is a single stateless instantiation and the builder pays one pointer
+    /// per requested relation, with no type-erasure allocation and no change to the builder's type -
+    /// which keeps the fluent chain and the asynchronous execution mode working unchanged.
+    using RelationPreloader = void (*)(DataMapper&, std::span<Record* const>);
+    std::vector<RelationPreloader> _relationPreloaders;
+
     friend class SqlWhereClauseBuilder<Derived>;
 
     LIGHTWEIGHT_FORCE_INLINE SqlSearchCondition& SearchCondition() noexcept
@@ -100,6 +132,50 @@ class [[nodiscard]] SqlCoreDataMapperQueryBuilder: public SqlBasicSelectQueryBui
         return RunFinisher([this] { return AllImpl(); });
     }
 
+    /// @brief Eagerly loads the given relation for every record the query returns, in a bounded
+    ///        number of queries instead of one per record.
+    ///
+    /// Touching a relation on a query result normally loads it on demand, one query per record - the
+    /// N+1 problem. `With<>()` instead resolves the relation for the whole result set after it has
+    /// been materialized: the keys are collected, the related rows are fetched with `WHERE ... IN
+    /// (...)` (chunked, see @ref SqlQueryFormatter::MaxInPredicateValues), and the rows are
+    /// distributed to the records in memory.
+    ///
+    /// Call it once per relation to load; the calls chain.
+    ///
+    /// Naming several relations forms a *path*: the first is resolved for the result set, the records
+    /// it loaded are then gathered and the next relation resolved for all of them at once. A path of
+    /// any length still costs a constant number of queries per level, never one per record - which is
+    /// what a nested `BelongsTo` needs, since each record holds its own copy of the target and
+    /// touching that copy's own relation would otherwise be an N+1 one level down.
+    ///
+    /// @tparam RelationPath One or more relations, in the form of `&Record::FieldName`. The first
+    ///         must be a member of the queried record, each subsequent one a member of the preceding
+    ///         relation's target. Supported for `BelongsTo` and `HasMany`; naming any other member is
+    ///         a compile error.
+    ///
+    /// @code
+    /// auto tracks = dm.Query<Track>()
+    ///                 .With<&Track::album>()                  // one extra SELECT ... WHERE id IN (...)
+    ///                 .With<&Track::album, &Album::artist>()  // one more, for every album at once
+    ///                 .All();
+    ///
+    /// for (auto& track: tracks)
+    ///     std::println("{} - {}", track.album.Record().title,
+    ///                  track.album.Record().artist.Record().name);  // no queries here
+    /// @endcode
+    ///
+    /// @note Relations that were not named keep their usual on-demand behaviour. Combining `With<>()`
+    ///       with `DataMapperOptions { .loadRelations = false }` therefore makes any *unrequested*
+    ///       relation throw `SqlRequireLoadedError` on access rather than quietly issuing a query.
+    ///       To eagerly load everything instead of naming paths, see
+    ///       @ref DataMapperOptions::eagerLoadDepth.
+    ///
+    /// @return This builder, for chaining.
+    template <auto... RelationPath>
+        requires DataMapperRecord<Record> && (sizeof...(RelationPath) >= 1)
+    [[nodiscard]] Derived& With();
+
     /// Executes a DELETE query.
     [[nodiscard]] auto Delete()
     {
@@ -227,6 +303,11 @@ class [[nodiscard]] SqlCoreDataMapperQueryBuilder: public SqlBasicSelectQueryBui
     // Synchronous implementations shared by both execution modes. The public finishers above
     // forward to these; the SQL building and result mapping live here exactly once.
 
+    /// Runs every loader registered through `With<>()` over the materialized result set.
+    ///
+    /// @param records The records to resolve the requested relations for.
+    void RunRelationPreloaders(std::span<Record> records);
+
     [[nodiscard]] bool ExistImpl();
     [[nodiscard]] size_t CountImpl();
     [[nodiscard]] std::vector<Record> AllImpl();
diff --git a/src/Lightweight/SqlQuery/Migrate.hpp b/src/Lightweight/SqlQuery/Migrate.hpp
index ab7861ca2..6fa4ba9e4 100644
--- a/src/Lightweight/SqlQuery/Migrate.hpp
+++ b/src/Lightweight/SqlQuery/Migrate.hpp
@@ -299,6 +299,11 @@ namespace detail
                         SqlForeignKeyReferenceDefinition {
                             .tableName = std::string { RecordTableName<typename FieldType::ReferencedRecord> },
                             .columnName = std::string { FieldNameOf<FieldType::ReferencedField> } });
+                    // A foreign key is the column every HasMany load filters on, and none of the
+                    // supported engines indexes one implicitly - only MySQL does. Without this the
+                    // relation query is a full table scan per owner, which measured 36x slower on
+                    // SQLite (6270 ms vs 174 ms for 1000 owners x 10 children).
+                    builder.Index();
                 }
                 else if constexpr (FieldType::IsMandatory)
                     builder.RequiredColumn(std::string(FieldNameOf<el>),
@@ -338,6 +343,9 @@ namespace detail
                             .tableName = std::string { RecordTableName<typename FieldType::ReferencedRecord> },
                             .columnName =
                                 std::string { FieldNameAt<referencedFieldIndex, typename FieldType::ReferencedRecord>() } });
+                    // See the comment in the reflection branch above: an unindexed foreign key turns
+                    // every HasMany load into a full table scan.
+                    builder.Index();
                 }
                 else if constexpr (FieldType::IsMandatory)
                     builder.RequiredColumn(std::string(FieldNameAt<I, Record>()),
diff --git a/src/Lightweight/SqlQueryFormatter.hpp b/src/Lightweight/SqlQueryFormatter.hpp
index 21426a7f5..99ec1b7ff 100644
--- a/src/Lightweight/SqlQueryFormatter.hpp
+++ b/src/Lightweight/SqlQueryFormatter.hpp
@@ -213,6 +213,23 @@ class [[nodiscard]] LIGHTWEIGHT_API SqlQueryFormatter
         return false;
     }
 
+    /// @brief Largest number of values this dialect should be handed in a single `WHERE ... IN (...)`
+    /// predicate.
+    ///
+    /// The batched relation loading behind `Query<Record>().With<&Record::relation>()` resolves a whole
+    /// result set through `IN` predicates over the collected keys. Those keys are rendered as literals,
+    /// so a large batch would otherwise produce one enormous expression — SQL Server raises
+    /// "an expression services limit has been reached", and SQLite has its own parser limits. Splitting
+    /// the batch into chunks of this size keeps every generated statement inside what the dialect
+    /// accepts; the cost is one extra round-trip per chunk, still a constant number of queries per
+    /// relation instead of one per record.
+    ///
+    /// @return The maximum number of values per `IN` predicate. Defaults to 1000 for every dialect.
+    [[nodiscard]] virtual size_t MaxInPredicateValues() const noexcept
+    {
+        return 1000;
+    }
+
     /// @brief Builds the canonical foreign-key constraint name for a set of columns.
     ///
     /// Produces `FK_<table>_<col1>[_<col2>…]`. A single-column FK collapses to
diff --git a/src/Lightweight/SqlStatement.cpp b/src/Lightweight/SqlStatement.cpp
index a3e8a79ac..1fc19d12e 100644
--- a/src/Lightweight/SqlStatement.cpp
+++ b/src/Lightweight/SqlStatement.cpp
@@ -9,6 +9,7 @@
 #include "Utils.hpp"
 
 #include <algorithm>
+#include <array>
 #include <cstddef>
 #include <cstdint>
 #include <cstring>
@@ -230,7 +231,9 @@ SqlStatement::SqlStatement(SqlStatement&& other) noexcept:
     m_connection { other.m_connection },
     m_hStmt { other.m_hStmt },
     m_preparedQuery { std::move(other.m_preparedQuery) },
-    m_expectedParameterCount { other.m_expectedParameterCount }
+    m_expectedParameterCount { other.m_expectedParameterCount },
+    m_preparedParameterCount { other.m_preparedParameterCount },
+    m_reusedPreparedQuery { other.m_reusedPreparedQuery }
 {
     other.m_data.reset();
     other.m_connection = nullptr;
@@ -247,6 +250,8 @@ SqlStatement& SqlStatement::operator=(SqlStatement&& other) noexcept
     m_hStmt = other.m_hStmt;
     m_preparedQuery = std::move(other.m_preparedQuery);
     m_expectedParameterCount = other.m_expectedParameterCount;
+    m_preparedParameterCount = other.m_preparedParameterCount;
+    m_reusedPreparedQuery = other.m_reusedPreparedQuery;
 
     other.m_data.reset();
     other.m_connection = nullptr;
@@ -291,7 +296,20 @@ void SqlStatement::Prepare(std::string_view query) &
     ZoneTextObject(query);
     SqlLogger::GetLogger().OnPrepare(query);
 
-    m_preparedQuery = std::string(query);
+    // Preparing the very same query text again is a wasted round-trip: the handle already holds that
+    // prepared statement. It is a common shape - one loader, one INSERT or one QuerySingle repeated
+    // per record - and the saving is dialect-dependent but large where it applies: measured over 1000
+    // identical single-row selects, skipping the re-prepare took PostgreSQL from 1053 ms to 190 ms
+    // (psqlODBC prepares server-side, one round-trip per call), while MS SQL Server and SQLite are
+    // unaffected because their drivers already fold it away.
+    //
+    // Everything below this decision still runs unconditionally: the handle carries column bindings,
+    // indicators, post-execute callbacks and possibly parameter-array attributes from the previous
+    // execution, and those must be torn down whether or not the query text changed.
+    bool const reusePreparedQuery = !m_preparedQuery.empty() && m_preparedQuery == query;
+    m_reusedPreparedQuery = reusePreparedQuery;
+    if (!reusePreparedQuery)
+        m_preparedQuery = std::string(query);
     const_cast<SqlStatement*>(this)->m_numColumns.reset();
 
     m_data->postExecuteCallbacks.clear();
@@ -315,10 +333,61 @@ void SqlStatement::Prepare(std::string_view query) &
     // but psqlODBC has historically treated SQL_C_CHAR parameter binds differently
     // depending on the variant of the most recent statement-text call — so we keep
     // the path uniformly W to side-step that.
-    auto wQuery = detail::OdbcWideArg { query };
+    if (!reusePreparedQuery)
+    {
+        auto wQuery = detail::OdbcWideArg { query };
+        RequireSuccess(SQLPrepareW(m_hStmt, wQuery.data(), static_cast<SQLINTEGER>(wQuery.buffer.size())));
+        RequireSuccess(SQLNumParams(m_hStmt, &m_expectedParameterCount));
+        m_preparedParameterCount = m_expectedParameterCount;
+    }
+    else
+        // SQLNumParams() was skipped, so restore what it would have reported. BindInputParameter()
+        // overwrites m_expectedParameterCount with SQLSMALLINT max to mean "the caller bound the
+        // parameters by hand"; leaving that in place would make the next Execute()/ExecuteBatch()
+        // reject its argument count (e.g. Create() followed by CreateAll(), which prepare byte-identical
+        // INSERT text), and would size the indicator vector to 32768 entries.
+        m_expectedParameterCount = m_preparedParameterCount;
+
+    m_data->indicators.resize(static_cast<size_t>(m_expectedParameterCount) + 1);
+}
+
+bool SqlStatement::RetryStalePreparedStatement(SQLRETURN result)
+{
+    if (SQL_SUCCEEDED(result) || result == SQL_NO_DATA || !m_reusedPreparedQuery)
+        return false;
+
+    // SQLSTATEs that mean "the prepared statement this handle holds can no longer be executed", as
+    // opposed to "this statement ran and the data was rejected". Only these are worth a second
+    // attempt: re-running a constraint violation would just fail again.
+    //
+    // - 42S02 The plan names a table that no longer resolves. MS SQL Server compiles a prepared
+    //         statement against the object *ids* it saw, so dropping and recreating a table between
+    //         two executions of the same handle invalidates the plan while the SQL text stays valid -
+    //         exactly the shape a schema migration produces.
+    // - 42P01 The PostgreSQL spelling of the same condition (`undefined_table`).
+    // - 0A000 PostgreSQL raises `cached plan must not change result type` when DDL changed a table
+    //         the cached plan reads.
+    // - 26000 The server no longer knows the prepared statement (e.g. after a `DISCARD ALL`).
+    // - 42P05 A duplicate prepared-statement name on the PostgreSQL side.
+    //
+    // Re-preparing a statement whose table is genuinely absent costs one extra round-trip and then
+    // reports the same error, so a real failure is still a failure.
+    static constexpr auto StalePreparedStatementStates =
+        std::array<std::string_view, 5> { "42S02", "42P01", "0A000", "26000", "42P05" };
+
+    auto const errorInfo = SqlErrorInfo::FromStatementHandle(m_hStmt);
+    if (!std::ranges::contains(StalePreparedStatementStates, std::string_view { errorInfo.sqlState }))
+        return false;
+
+    SqlLogger::GetLogger().OnWarning(std::format(
+        "Re-preparing statement after the server rejected the cached one ({}): {}", errorInfo.sqlState, m_preparedQuery));
+
+    auto wQuery = detail::OdbcWideArg { std::string_view { m_preparedQuery } };
     RequireSuccess(SQLPrepareW(m_hStmt, wQuery.data(), static_cast<SQLINTEGER>(wQuery.buffer.size())));
     RequireSuccess(SQLNumParams(m_hStmt, &m_expectedParameterCount));
-    m_data->indicators.resize(static_cast<size_t>(m_expectedParameterCount) + 1);
+    m_preparedParameterCount = m_expectedParameterCount;
+    m_reusedPreparedQuery = false;
+    return true;
 }
 
 SqlResultCursor SqlStatement::ExecuteDirect(std::string_view const& query, std::source_location location)
@@ -329,6 +398,7 @@ SqlResultCursor SqlStatement::ExecuteDirect(std::string_view const& query, std::
         return SqlResultCursor { *this };
 
     m_preparedQuery.clear();
+    m_reusedPreparedQuery = false;
     m_numColumns.reset();
 
     RequireSuccess(SQLFreeStmt(m_hStmt, SQL_UNBIND));
@@ -372,7 +442,12 @@ SqlResultCursor SqlStatement::ExecuteWithVariants(std::vector<SqlVariant> const&
         SqlDataBinder<SqlVariant>::InputParameter(m_hStmt, static_cast<SQLUSMALLINT>(1 + i), arg, *this);
     }
 
-    auto const rc = SQLExecute(m_hStmt);
+    auto rc = SQLExecute(m_hStmt);
+
+    // A prepared statement Prepare() reused rather than re-issued can have gone stale server-side.
+    if (RetryStalePreparedStatement(rc))
+        rc = SQLExecute(m_hStmt);
+
     if (rc != SQL_NO_DATA)
         RequireSuccess(rc);
     ProcessPostExecuteCallbacks();
@@ -420,6 +495,7 @@ RowArrayCursor SqlStatement::ExecuteBatchFetch(std::string_view query, std::size
         throw std::invalid_argument { "arrayDepth must be greater than zero" };
 
     m_preparedQuery.clear();
+    m_reusedPreparedQuery = false;
     m_numColumns.reset();
 
     RequireSuccess(SQLFreeStmt(m_hStmt, SQL_UNBIND));
diff --git a/src/Lightweight/SqlStatement.hpp b/src/Lightweight/SqlStatement.hpp
index 9da74479f..c3945443f 100644
--- a/src/Lightweight/SqlStatement.hpp
+++ b/src/Lightweight/SqlStatement.hpp
@@ -391,6 +391,24 @@ class [[nodiscard]] SqlStatement final: public SqlDataBinderCallback
         SQLULEN processedCount,
         SQLULEN expectedCount,
         std::source_location sourceLocation = std::source_location::current()) const;
+    /// @brief Re-issues @c SQLPrepareW after a reused prepared statement went stale, for one retry.
+    ///
+    /// @c Prepare() skips @c SQLPrepareW when the statement handle already holds exactly this query
+    /// text. A prepared statement can nevertheless stop being executable - most visibly on PostgreSQL,
+    /// where a DDL change between two executions makes the server reject the cached plan with
+    /// @c 0A000 - so an execute that fails with one of those SQLSTATEs re-prepares once and runs again.
+    /// Parameter bindings live on the handle and survive @c SQLPrepareW, so the caller's already-bound
+    /// arguments stay valid for the retry.
+    ///
+    /// @note Retrying re-executes the whole statement. That is safe for the conditions listed above
+    ///       because they are all raised while resolving or planning the statement, before it has had
+    ///       any effect - a constraint violation or any other runtime rejection is deliberately not
+    ///       retried.
+    ///
+    /// @param result The @c SQLRETURN of the execute that just failed.
+    /// @return @c true if the statement was re-prepared and the execute should be retried.
+    [[nodiscard]] LIGHTWEIGHT_API bool RetryStalePreparedStatement(SQLRETURN result);
+
     LIGHTWEIGHT_API void RequireIndicators();
     LIGHTWEIGHT_API SQLLEN* GetIndicatorForColumn(SQLUSMALLINT column) noexcept;
 
@@ -460,6 +478,12 @@ class [[nodiscard]] SqlStatement final: public SqlDataBinderCallback
     std::string m_preparedQuery;                   // The last prepared query
     std::optional<SQLSMALLINT> m_numColumns;       // The number of columns in the result set, if known
     SQLSMALLINT m_expectedParameterCount {};       // The number of parameters expected by the query
+    // The parameter count SQLNumParams() reported for m_preparedQuery. m_expectedParameterCount is
+    // deliberately overwritten by BindInputParameter() (with SQLSMALLINT max, meaning "bound by hand"),
+    // so a Prepare() that reuses the handle's statement - and therefore skips SQLNumParams() - has to
+    // restore the real count from here instead.
+    SQLSMALLINT m_preparedParameterCount {};
+    bool m_reusedPreparedQuery { false }; // Whether the last Prepare() reused the handle's statement
 };
 
 /// @ingroup CoreApi
@@ -1164,7 +1188,11 @@ SqlResultCursor SqlStatement::Execute(Args const&... args)
       RequireSuccess(SqlDataBinder<Args>::InputParameter(m_hStmt, i, args, *this))),
      ...);
 
-    auto const result = SQLExecute(m_hStmt);
+    auto result = SQLExecute(m_hStmt);
+
+    // A prepared statement Prepare() reused rather than re-issued can have gone stale server-side.
+    if (RetryStalePreparedStatement(result))
+        result = SQLExecute(m_hStmt);
 
     if (result != SQL_NO_DATA && result != SQL_SUCCESS && result != SQL_SUCCESS_WITH_INFO)
         throw SqlException(SqlErrorInfo::FromStatementHandle(m_hStmt), std::source_location::current());
diff --git a/src/benchmark/CMakeLists.txt b/src/benchmark/CMakeLists.txt
index 4e3690195..c36734ef2 100644
--- a/src/benchmark/CMakeLists.txt
+++ b/src/benchmark/CMakeLists.txt
@@ -10,6 +10,13 @@ if(EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/entities/CMakeLists.txt)
     target_link_libraries(LightweightBenchmark LightweightEntities)
 endif()
 
+# Runtime benchmark for relation loading (issue #563): measures wall-clock time and statement counts
+# of the on-demand loaders against `With<>()` eager loading. Needs a live database to run, but builds
+# without one.
+add_executable(LightweightRelationBenchmark relations_runtime.cpp)
+target_compile_features(LightweightRelationBenchmark PUBLIC cxx_std_23)
+target_link_libraries(LightweightRelationBenchmark Lightweight::Lightweight)
+
 # Compilation-time profiling. When enabled (and building with Clang), emit a
 # Chrome-trace (`benchmark.cpp.json`) next to the object file that
 # measure_compile_time.py aggregates into per-template-instantiation timings.
diff --git a/src/benchmark/relations_runtime.cpp b/src/benchmark/relations_runtime.cpp
new file mode 100644
index 000000000..a9d86e042
--- /dev/null
+++ b/src/benchmark/relations_runtime.cpp
@@ -0,0 +1,303 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// Runtime benchmark for relation loading (issue #563).
+//
+// Unlike `benchmark.cpp`, which measures *compile* time, this one measures wall-clock time and the
+// number of statements issued, comparing the on-demand relation loaders against `With<>()` eager
+// loading over the same data. It needs a live database: pass an ODBC connection string as the third
+// argument to run it against something other than the default local SQLite file.
+//
+//     LightweightRelationBenchmark [owners] [childrenPerOwner] [connectionString]
+//
+// The statement counts come from a SqlLogger subclass rather than from an estimate, so a change that
+// silently reintroduces an N+1 shows up as a count, not just as a slower number.
+
+#include <Lightweight/Lightweight.hpp>
+
+#include <algorithm>
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+using namespace Lightweight;
+using namespace std::chrono;
+
+#if defined(LIGHTWEIGHT_CXX26_REFLECTION)
+    #define Member(x) ^^x
+#else
+    #define Member(x) &x
+#endif
+
+struct Child;
+
+struct Owner
+{
+    static constexpr std::string_view TableName = "BenchOwner";
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<40>> name {};
+    HasMany<Child> children {};
+};
+
+struct Child
+{
+    static constexpr std::string_view TableName = "BenchChild";
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<40>> label {};
+    BelongsTo<Member(Owner::id), SqlRealName { "owner_id" }> owner {};
+};
+
+// Counts prepares/executes so the query count per strategy is reported, not guessed.
+class CountingLogger final: public SqlLogger::Null
+{
+  public:
+    size_t prepares = 0;
+    size_t executes = 0;
+    size_t directs = 0;
+    void OnPrepare(std::string_view const&) override
+    {
+        ++prepares;
+    }
+    void OnExecute(std::string_view const&) override
+    {
+        ++executes;
+    }
+    void OnExecuteDirect(std::string_view const&) override
+    {
+        ++directs;
+    }
+    void Reset()
+    {
+        prepares = executes = directs = 0;
+    }
+};
+
+namespace
+{
+CountingLogger g_logger;
+
+size_t g_repetitions = 3;
+
+/// Runs @p f g_repetitions times and returns the fastest wall-clock time, so a cold connection,
+/// a cold page cache or a one-off server-side plan compilation does not get reported as the cost
+/// of the strategy under test.
+template <typename F>
+double TimeMs(F&& f)
+{
+    double best = 1e18;
+    for (size_t i = 0; i < g_repetitions; ++i)
+    {
+        auto const start = steady_clock::now();
+        f();
+        best = std::min(best, duration<double, std::milli>(steady_clock::now() - start).count());
+        if (i + 1 < g_repetitions)
+            g_logger.Reset();
+    }
+    return best;
+}
+
+void Report(char const* name, double ms, size_t rows)
+{
+    std::printf(
+        "%-46s %9.2f ms   prepare=%-6zu execute=%-6zu rows=%zu\n", name, ms, g_logger.prepares, g_logger.executes, rows);
+    g_logger.Reset();
+}
+} // namespace
+
+int main(int argc, char** argv)
+{
+    size_t const owners = argc > 1 ? std::stoul(argv[1]) : 500;
+    size_t const childrenPerOwner = argc > 2 ? std::stoul(argv[2]) : 10;
+    std::string const connectionString = argc > 3 ? argv[3] : std::string { "DRIVER=SQLite3;Database=/tmp/lw-bench.sqlite" };
+
+    SqlConnection::SetDefaultConnectionString(SqlConnectionString { connectionString });
+    SqlLogger::SetLogger(g_logger);
+
+    auto dm = DataMapper {};
+    std::ignore = SqlStatement { dm.Connection() }.ExecuteDirect("DROP TABLE IF EXISTS \"BenchChild\"");
+    std::ignore = SqlStatement { dm.Connection() }.ExecuteDirect("DROP TABLE IF EXISTS \"BenchOwner\"");
+    dm.CreateTable<Owner>();
+    dm.CreateTable<Child>();
+
+    {
+        auto transaction = SqlTransaction { dm.Connection() };
+        for (size_t i = 0; i < owners; ++i)
+        {
+            auto owner = Owner { .name = SqlAnsiString<40> { std::format("owner-{}", i) } };
+            dm.Create(owner);
+            for (size_t j = 0; j < childrenPerOwner; ++j)
+            {
+                auto child = Child { .label = SqlAnsiString<40> { std::format("child-{}-{}", i, j) } };
+                child.owner = owner;
+                dm.Create(child);
+            }
+        }
+        transaction.Commit();
+    }
+    // Optional: index the foreign key column. CreateTable() emits a FOREIGN KEY constraint but no
+    // index, and none of the three supported engines indexes a foreign key implicitly - so this
+    // isolates "N queries" from "N full table scans".
+    if (char const* const withIndex = std::getenv("LW_BENCH_INDEX"); withIndex && *withIndex == '1')
+        std::ignore =
+            SqlStatement { dm.Connection() }.ExecuteDirect(R"(CREATE INDEX "ix_bench_owner" ON "BenchChild" ("owner_id"))");
+
+    // Warm-up: opens the connection's first cursor and warms the server-side caches.
+    (void) dm.Query<Owner, DataMapperOptions { .loadRelations = false }>().All();
+    g_logger.Reset();
+
+    std::printf("== %zu owners x %zu children (%s)\n", owners, childrenPerOwner, connectionString.c_str());
+
+    // (1) Baseline: parents only, relations auto-loading configured (the default).
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] { rows = dm.Query<Owner>().All().size(); });
+        Report("parents only, loadRelations=true (default)", ms, rows);
+    }
+
+    // (2) Baseline: parents only, no loader installation at all.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] { rows = dm.Query<Owner, DataMapperOptions { .loadRelations = false }>().All().size(); });
+        Report("parents only, loadRelations=false", ms, rows);
+    }
+
+    // (3) The N+1: touch HasMany on every parent.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto parents = dm.Query<Owner>().All();
+            for (auto& p: parents)
+                rows += p.children.All().size();
+        });
+        Report("HasMany lazy per parent (N+1)", ms, rows);
+    }
+
+    // (4) The fix, done by hand: one extra SELECT ... WHERE owner_id IN (...) + in-memory stitch.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto parents = dm.Query<Owner, DataMapperOptions { .loadRelations = false }>().All();
+            std::vector<uint64_t> ids;
+            ids.reserve(parents.size());
+            for (auto const& p: parents)
+                ids.push_back(p.id.Value());
+            auto kids = dm.Query<Child, DataMapperOptions { .loadRelations = false }>()
+                            .WhereIn(FieldNameOf<Member(Child::owner)>, ids)
+                            .All();
+            std::unordered_map<uint64_t, std::vector<std::shared_ptr<Child>>> byOwner;
+            byOwner.reserve(parents.size());
+            for (auto& k: kids)
+                byOwner[k.owner.Value()].push_back(std::make_shared<Child>(std::move(k)));
+            for (auto& p: parents)
+            {
+                auto it = byOwner.find(p.id.Value());
+                if (it != byOwner.end())
+                    rows += p.children.Emplace(std::move(it->second)).size();
+            }
+        });
+        Report("HasMany batched WhereIn + stitch (2 queries)", ms, rows);
+    }
+
+    // (5) BelongsTo N+1: touch the parent of every child.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto kids = dm.Query<Child>().All();
+            for (auto& k: kids)
+                rows += k.owner.Record().id.Value() != 0 ? 1 : 0;
+        });
+        Report("BelongsTo lazy per child (N+1)", ms, rows);
+    }
+
+    // (6) BelongsTo batched: distinct FK values, one IN query, distribute.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto kids = dm.Query<Child, DataMapperOptions { .loadRelations = false }>().All();
+            std::vector<uint64_t> ids;
+            ids.reserve(kids.size());
+            for (auto const& k: kids)
+                ids.push_back(k.owner.Value());
+            std::ranges::sort(ids);
+            ids.erase(std::ranges::unique(ids).begin(), ids.end());
+            auto parents = dm.Query<Owner, DataMapperOptions { .loadRelations = false }>()
+                               .WhereIn(FieldNameOf<Member(Owner::id)>, ids)
+                               .All();
+            std::unordered_map<uint64_t, Owner*> byId;
+            byId.reserve(parents.size());
+            for (auto& p: parents)
+                byId[p.id.Value()] = &p;
+            for (auto& k: kids)
+            {
+                auto it = byId.find(k.owner.Value());
+                if (it != byId.end())
+                {
+                    k.owner.AdoptFetchedRecord(*it->second);
+                    ++rows;
+                }
+            }
+        });
+        Report("BelongsTo batched WhereIn + stitch (2 queries)", ms, rows);
+    }
+
+    // (7)/(8) How much of one lazy load is the re-Prepare? The lazy loaders call
+    // SqlStatement::Prepare() on identical SQL for every owner; SqlStatement has no
+    // prepared-statement cache, so each iteration re-issues SQLPrepareW.
+    {
+        auto const sql = std::format(R"(SELECT "id", "label", "owner_id" FROM "BenchChild" WHERE "owner_id" = ?)");
+        size_t rows = 0;
+        auto stmt = SqlStatement { dm.Connection() };
+        auto const ms = TimeMs([&] {
+            for (size_t i = 1; i <= owners; ++i)
+            {
+                stmt.Prepare(sql);
+                auto cursor = stmt.Execute(static_cast<uint64_t>(i));
+                while (cursor.FetchRow())
+                    ++rows;
+            }
+        });
+        Report("raw: Prepare+Execute per owner (N queries)", ms, rows);
+    }
+    {
+        auto const sql = std::format(R"(SELECT "id", "label", "owner_id" FROM "BenchChild" WHERE "owner_id" = ?)");
+        size_t rows = 0;
+        auto stmt = SqlStatement { dm.Connection() };
+        auto const ms = TimeMs([&] {
+            stmt.Prepare(sql);
+            for (size_t i = 1; i <= owners; ++i)
+            {
+                auto cursor = stmt.Execute(static_cast<uint64_t>(i));
+                while (cursor.FetchRow())
+                    ++rows;
+            }
+        });
+        Report("raw: Prepare once + Execute per owner", ms, rows);
+    }
+
+    // (9) The new API: HasMany eager-loaded for the whole batch.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto parents = dm.Query<Owner>().With<Member(Owner::children)>().All();
+            for (auto& p: parents)
+                rows += p.children.All().size();
+        });
+        Report("HasMany .With<>() eager", ms, rows);
+    }
+
+    // (10) The new API: BelongsTo eager-loaded for the whole batch.
+    {
+        size_t rows = 0;
+        auto const ms = TimeMs([&] {
+            auto kids = dm.Query<Child>().With<Member(Child::owner)>().All();
+            for (auto& k: kids)
+                rows += k.owner.Record().id.Value() != 0 ? 1 : 0;
+        });
+        Report("BelongsTo .With<>() eager", ms, rows);
+    }
+
+    return 0;
+}
diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt
index 8cc243a85..b7879657b 100644
--- a/src/tests/CMakeLists.txt
+++ b/src/tests/CMakeLists.txt
@@ -47,6 +47,7 @@ set(SOURCE_FILES
     SecretResolverTests.cpp
     DataMapper/BatchTests.cpp
     DataMapper/CreateTests.cpp
+    DataMapper/EagerLoadingTests.cpp
     DataMapper/MiscTests.cpp
     DataMapper/NamingTests.cpp
     DataMapper/ReadTests.cpp
diff --git a/src/tests/DataMapper/BatchTests.cpp b/src/tests/DataMapper/BatchTests.cpp
index 3f1ce3093..11b1af444 100644
--- a/src/tests/DataMapper/BatchTests.cpp
+++ b/src/tests/DataMapper/BatchTests.cpp
@@ -172,6 +172,28 @@ TEST_CASE_METHOD(SqlTestFixture, "DataMapper.CreateAll: empty and single span",
     }
 }
 
+TEST_CASE_METHOD(SqlTestFixture,
+                 "DataMapper.CreateAll: after a single Create of the same record type",
+                 "[DataMapper][batch]")
+{
+    // Create() prepares byte-identical INSERT text to CreateAll() and binds its parameters by hand,
+    // which marks the statement's parameter count as "bound by the caller". A Prepare() that reuses
+    // the statement already on the handle skips SQLNumParams(), so that marker has to be undone
+    // explicitly - otherwise the batch execute below rejects its column count.
+    auto dm = DataMapper {};
+    dm.CreateTable<BatchFixedRecord>();
+
+    auto single = BatchFixedRecord { .id = 1, .value = 1.5, .count = 10 };
+    dm.Create(single);
+
+    auto records = std::vector<BatchFixedRecord> {};
+    for (auto const i: std::views::iota(2, 6))
+        records.push_back({ .id = i, .value = i * 1.5, .count = i * 10 });
+    dm.CreateAll(records);
+
+    CHECK(dm.Query<BatchFixedRecord>().Count() == 5);
+}
+
 TEST_CASE_METHOD(SqlTestFixture, "DataMapper.UpdateAll: writes all columns incl. NULL", "[DataMapper][batch]")
 {
     auto dm = DataMapper {};
diff --git a/src/tests/DataMapper/EagerLoadingTests.cpp b/src/tests/DataMapper/EagerLoadingTests.cpp
new file mode 100644
index 000000000..ae931d64c
--- /dev/null
+++ b/src/tests/DataMapper/EagerLoadingTests.cpp
@@ -0,0 +1,581 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// Eager relation loading: `Query<Record>().With<&Record::relation>()`.
+//
+// The point of the feature is the *number of queries*, so correctness alone is not enough to test:
+// a `With<>()` that quietly fell back to the on-demand loaders would still return the right records
+// while reintroducing exactly the N+1 the feature exists to remove. Every test here therefore
+// asserts the statement count through a counting `SqlLogger` alongside the data.
+
+#include "../Utils.hpp"
+
+#include <Lightweight/DataMapper/DataMapper.hpp>
+#include <Lightweight/SqlLogger.hpp>
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <cstddef>
+#include <format>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+
+using namespace Lightweight;
+
+struct EagerChild;
+
+struct EagerCategory
+{
+    static constexpr std::string_view TableName = "EagerCategory";
+
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<32>> title {};
+};
+
+struct EagerRegion
+{
+    static constexpr std::string_view TableName = "EagerRegion";
+
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<32>> label {};
+};
+
+struct EagerOwner
+{
+    static constexpr std::string_view TableName = "EagerOwner";
+
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<32>> name {};
+    BelongsTo<Member(EagerRegion::id), SqlRealName { "region_id" }> region {};
+    HasMany<EagerChild> children {};
+};
+
+struct EagerChild
+{
+    static constexpr std::string_view TableName = "EagerChild";
+
+    Field<uint64_t, PrimaryKey::ServerSideAutoIncrement> id {};
+    Field<SqlAnsiString<32>> label {};
+    BelongsTo<Member(EagerOwner::id), SqlRealName { "owner_id" }> owner {};
+    BelongsTo<Member(EagerCategory::id), SqlRealName { "category_id" }, SqlNullable::Null> category {};
+};
+
+namespace
+{
+
+/// Counts the statements the mapper issues, so a test can assert "two queries" rather than trusting
+/// that the batched path was taken.
+///
+/// Only `OnPrepare` and `OnExecuteDirect` are counted: every query the mapper runs passes through one
+/// of the two, while `OnExecute` would double-count a prepared statement.
+class StatementCountingLogger final: public SqlLogger::Null
+{
+  public:
+    void OnPrepare(std::string_view const& /*query*/) override
+    {
+        ++_statements;
+    }
+
+    void OnExecuteDirect(std::string_view const& /*query*/) override
+    {
+        ++_statements;
+    }
+
+    /// Statements issued since the last Reset().
+    [[nodiscard]] size_t Count() const noexcept
+    {
+        return _statements;
+    }
+
+    void Reset() noexcept
+    {
+        _statements = 0;
+    }
+
+  private:
+    size_t _statements = 0;
+};
+
+/// Installs a statement-counting logger for the duration of a test, restoring the previous one.
+class ScopedStatementCounter
+{
+  public:
+    ScopedStatementCounter():
+        _previous { &SqlLogger::GetLogger() }
+    {
+        SqlLogger::SetLogger(_counter);
+    }
+
+    ScopedStatementCounter(ScopedStatementCounter const&) = delete;
+    ScopedStatementCounter(ScopedStatementCounter&&) = delete;
+    ScopedStatementCounter& operator=(ScopedStatementCounter const&) = delete;
+    ScopedStatementCounter& operator=(ScopedStatementCounter&&) = delete;
+
+    ~ScopedStatementCounter()
+    {
+        SqlLogger::SetLogger(*_previous);
+    }
+
+    /// Statements issued since construction or the last Reset().
+    [[nodiscard]] size_t Count() const noexcept
+    {
+        return _counter.Count();
+    }
+
+    void Reset() noexcept
+    {
+        _counter.Reset();
+    }
+
+  private:
+    StatementCountingLogger _counter;
+    SqlLogger* _previous;
+};
+
+/// Creates the two tables and fills them with @p ownerCount owners, each owning @p childrenPerOwner
+/// children.
+///
+/// @param dm The mapper to create the schema and rows through.
+/// @param ownerCount How many owner rows to create.
+/// @param childrenPerOwner How many children each owner gets.
+void MakeOwnersWithChildren(DataMapper& dm, size_t ownerCount, size_t childrenPerOwner)
+{
+    dm.CreateTable<EagerCategory>();
+    dm.CreateTable<EagerRegion>();
+    dm.CreateTable<EagerOwner>();
+    dm.CreateTable<EagerChild>();
+
+    auto category = EagerCategory { .title = "shared" };
+    dm.Create(category);
+
+    auto region = EagerRegion { .label = "north" };
+    dm.Create(region);
+
+    for (size_t ownerIndex = 0; ownerIndex < ownerCount; ++ownerIndex)
+    {
+        auto owner = EagerOwner { .name = SqlAnsiString<32> { std::format("owner-{}", ownerIndex) } };
+        owner.region = region;
+        dm.Create(owner);
+
+        for (size_t childIndex = 0; childIndex < childrenPerOwner; ++childIndex)
+        {
+            auto child = EagerChild { .label = SqlAnsiString<32> { std::format("child-{}-{}", ownerIndex, childIndex) } };
+            child.owner = owner;
+            // Every other child leaves the optional relation NULL, so the batched loader has to cope
+            // with a foreign key that references nothing.
+            if (childIndex % 2 == 0)
+                child.category = category;
+            dm.Create(child);
+        }
+    }
+}
+
+} // namespace
+
+TEST_CASE_METHOD(SqlTestFixture, "With<HasMany> loads every owner's children in one extra query", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 5, 3);
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>().With<Member(EagerOwner::children)>().All();
+
+    // One SELECT for the owners, one for all their children - regardless of how many owners there are.
+    CHECK(counter.Count() == 2);
+
+    REQUIRE(owners.size() == 5);
+    for (auto& owner: owners)
+    {
+        CHECK(owner.children.Count() == 3);
+        for (auto const& child: owner.children.All())
+            CHECK(child->owner.Value() == owner.id.Value());
+    }
+
+    // Touching the relation must not have gone back to the database at all.
+    CHECK(counter.Count() == 2);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<BelongsTo> loads every child's owner in one extra query", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 4, 3);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild>().With<Member(EagerChild::owner)>().All();
+
+    CHECK(counter.Count() == 2);
+
+    REQUIRE(children.size() == 12);
+    for (auto& child: children)
+    {
+        CHECK(child.owner.Record().id.Value() == child.owner.Value());
+        CHECK(child.owner.Record().name.Value().ToStringView().starts_with("owner-"));
+    }
+
+    CHECK(counter.Count() == 2);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> chains across several relations", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 3, 2);
+
+    auto counter = ScopedStatementCounter {};
+    // Two relations named means two extra statements - not one per record, and not one per relation
+    // per record.
+    auto children = dm.Query<EagerChild>().With<Member(EagerChild::owner)>().With<Member(EagerChild::category)>().All();
+
+    CHECK(counter.Count() == 3);
+    REQUIRE(children.size() == 6);
+    for (auto& child: children)
+        CHECK(child.owner.Record().id.Value() == child.owner.Value());
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<BelongsTo> skips rows whose foreign key is NULL", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 3, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild>().With<Member(EagerChild::category)>().All();
+
+    CHECK(counter.Count() == 2);
+    REQUIRE(children.size() == 6);
+
+    size_t withCategory = 0;
+    for (auto& child: children)
+    {
+        if (!child.category.Value().has_value())
+            continue;
+        ++withCategory;
+
+        // A nullable BelongsTo hands back an optional reference; throwing rather than CHECK-ing makes
+        // the dereference below provably safe (a Catch2 assertion is a loop the analyser cannot follow).
+        auto const category = child.category.Record();
+        if (!category.has_value())
+            throw std::runtime_error("The eagerly loaded category must be present.");
+        CHECK(category->get().title.Value().ToStringView() == "shared");
+    }
+    // Half the children carry the optional relation; the loader must have fetched the one shared
+    // category exactly once for all of them.
+    CHECK(withCategory == 3);
+    CHECK(counter.Count() == 2);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<HasMany> marks childless owners loaded-empty", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 2, 2);
+
+    // An owner with no children at all. It still needs its (mandatory) region, so reuse the one the
+    // fixture created rather than tripping the foreign key.
+    auto const region = dm.Query<EagerRegion>().First();
+    if (!region.has_value())
+        throw std::runtime_error("The fixture must have created a region.");
+    auto lonely = EagerOwner { .name = "lonely" };
+    lonely.region = *region;
+    dm.Create(lonely);
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>().With<Member(EagerOwner::children)>().All();
+    REQUIRE(owners.size() == 3);
+
+    auto const it =
+        std::ranges::find_if(owners, [](EagerOwner const& owner) { return owner.name.Value().ToStringView() == "lonely"; });
+    REQUIRE(it != owners.end());
+
+    // The empty result is already known, so asking for it must not produce a query.
+    CHECK(it->children.IsEmpty());
+    CHECK(it->children.Count() == 0);
+    CHECK(counter.Count() == 2);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> applies to First(n) and Range() too", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 6, 2);
+
+    {
+        auto counter = ScopedStatementCounter {};
+        auto owners = dm.Query<EagerOwner>()
+                          .OrderBy(FieldNameOf<Member(EagerOwner::id)>, SqlResultOrdering::ASCENDING)
+                          .With<Member(EagerOwner::children)>()
+                          .First(3);
+        CHECK(counter.Count() == 2);
+        REQUIRE(owners.size() == 3);
+        for (auto& owner: owners)
+            CHECK(owner.children.Count() == 2);
+    }
+
+    {
+        auto counter = ScopedStatementCounter {};
+        auto owners = dm.Query<EagerOwner>()
+                          .OrderBy(FieldNameOf<Member(EagerOwner::id)>, SqlResultOrdering::ASCENDING)
+                          .With<Member(EagerOwner::children)>()
+                          .Range(2, 2);
+        CHECK(counter.Count() == 2);
+        REQUIRE(owners.size() == 2);
+        for (auto& owner: owners)
+            CHECK(owner.children.Count() == 2);
+    }
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> on an empty result set issues no relation query", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 0, 0);
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>().With<Member(EagerOwner::children)>().All();
+
+    CHECK(owners.empty());
+    CHECK(counter.Count() == 1);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> survives a batch larger than one IN chunk", "[DataMapper][With]")
+{
+    // The IN predicate is chunked (SqlQueryFormatter::MaxInPredicateValues), and the chunk boundary is
+    // where a batched loader most easily loses rows: the last partial chunk, or an owner whose key
+    // lands in a later chunk than the first.
+    auto dm = DataMapper {};
+    auto const ownerCount = size_t { 1005 };
+    MakeOwnersWithChildren(dm, ownerCount, 1);
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>().With<Member(EagerOwner::children)>().All();
+
+    REQUIRE(owners.size() == ownerCount);
+
+    // 1 for the owners + ceil(1005 / 1000) for the children: a constant per chunk, never per record.
+    CHECK(counter.Count() == 3);
+
+    size_t childrenSeen = 0;
+    for (auto& owner: owners)
+        childrenSeen += owner.children.Count();
+    CHECK(childrenSeen == ownerCount);
+    CHECK(counter.Count() == 3);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "Relations not named by With<> keep loading on demand", "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 3, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild>().All();
+    CHECK(counter.Count() == 1);
+
+    // No With<> was requested, so the first access still goes to the database - the feature must not
+    // change what an unrequested relation does.
+    CHECK(children.front().owner.Record().id.Value() != 0);
+    CHECK(counter.Count() > 1);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> walks a nested BelongsTo path in one query per level", "[DataMapper][With]")
+{
+    // The shape the one-level version cannot help with: every child holds its *own copy* of its
+    // owner, so reaching owner.region through the copy would run that copy's own lazy loader - one
+    // query per child, the N+1 moved one level down.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 6, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild>()
+                        .With<Member(EagerChild::owner)>()
+                        .With<Member(EagerChild::owner), Member(EagerOwner::region)>()
+                        .All();
+
+    // Children, their owners, the owners' regions: three statements for 12 children.
+    CHECK(counter.Count() == 3);
+
+    REQUIRE(children.size() == 12);
+    for (auto& child: children)
+    {
+        auto const& owner = child.owner.Record();
+        CHECK(owner.id.Value() == child.owner.Value());
+        CHECK(owner.region.Record().label.Value().ToStringView() == "north");
+    }
+
+    // Reading the whole nested graph must not have gone back to the database.
+    CHECK(counter.Count() == 3);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "With<> walks a path through a HasMany", "[DataMapper][With]")
+{
+    // Path through the "many" side: owners, their children, and each child's category - the middle
+    // level fans out, so a per-record walk here would be one query per *child*, not per owner.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 4, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>()
+                      .With<Member(EagerOwner::children)>()
+                      .With<Member(EagerOwner::children), Member(EagerChild::category)>()
+                      .All();
+
+    CHECK(counter.Count() == 3);
+    REQUIRE(owners.size() == 4);
+
+    size_t categorized = 0;
+    for (auto& owner: owners)
+        for (auto const& child: owner.children.All())
+            if (child->category.Value().has_value())
+            {
+                auto const category = child->category.Record();
+                if (!category.has_value())
+                    throw std::runtime_error("The eagerly loaded category must be present.");
+                CHECK(category->get().title.Value().ToStringView() == "shared");
+                ++categorized;
+            }
+
+    CHECK(categorized == 4);
+    CHECK(counter.Count() == 3);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "eagerLoadDepth loads every relation of the result set", "[DataMapper][With]")
+{
+    // No relation named at all: the option loads whatever is reachable, which for a child is its
+    // owner and its category (depth 1).
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 5, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild, DataMapperOptions { .eagerLoadDepth = 1 }>().All();
+
+    // Children + owners + categories.
+    CHECK(counter.Count() == 3);
+    REQUIRE(children.size() == 10);
+
+    for (auto& child: children)
+        CHECK(child.owner.Record().id.Value() == child.owner.Value());
+
+    CHECK(counter.Count() == 3);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "eagerLoadDepth descends through the relation graph", "[DataMapper][With]")
+{
+    // Depth 2 from a child reaches the owner's own relations as well: owner.region and, back down
+    // the inverse side, the owner's children.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 4, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild, DataMapperOptions { .eagerLoadDepth = 2 }>().All();
+
+    REQUIRE(children.size() == 8);
+
+    for (auto& child: children)
+    {
+        auto const& owner = child.owner.Record();
+        CHECK(owner.region.Record().label.Value().ToStringView() == "north");
+        CHECK(owner.children.Count() == 2);
+    }
+
+    // Whatever the exact number of levels walked, it must be a small constant - not one query per
+    // record, which for 8 children with two relations each would be well past twenty.
+    CHECK(counter.Count() <= 8);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "A named path is not re-fetched by eagerLoadDepth", "[DataMapper][With]")
+{
+    // Both mechanisms at once: the batched loaders skip a relation that is already in memory, so
+    // naming a path and asking for a depth walk must not query the same relation twice.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 3, 2);
+
+    auto counter = ScopedStatementCounter {};
+    auto children =
+        dm.Query<EagerChild, DataMapperOptions { .eagerLoadDepth = 1 }>().With<Member(EagerChild::owner)>().All();
+
+    // Children, owners (named), categories (from the depth walk) - the owners are not fetched twice.
+    CHECK(counter.Count() == 3);
+    REQUIRE(children.size() == 6);
+    for (auto& child: children)
+        CHECK(child.owner.Record().id.Value() == child.owner.Value());
+}
+
+TEST_CASE_METHOD(SqlTestFixture,
+                 "With<BelongsTo> issues no relation query when every foreign key is NULL",
+                 "[DataMapper][With]")
+{
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 0, 0); // schema only; the rows below leave `category` unset
+
+    auto region = EagerRegion { .label = "north" };
+    dm.Create(region);
+
+    auto owner = EagerOwner { .name = "solo" };
+    owner.region = region;
+    dm.Create(owner);
+
+    for (auto const index: { 0, 1 })
+    {
+        auto child = EagerChild { .label = SqlAnsiString<32> { std::format("orphan-{}", index) } };
+        child.owner = owner;
+        dm.Create(child);
+    }
+
+    auto counter = ScopedStatementCounter {};
+    auto children = dm.Query<EagerChild>().With<Member(EagerChild::category)>().All();
+
+    REQUIRE(children.size() == 2);
+    // There are rows, but not a single key to look up: the loader must stop after collecting targets
+    // instead of querying the category table for nothing.
+    CHECK(counter.Count() == 1);
+    for (auto const& child: children)
+        CHECK_FALSE(child.category.Value().has_value());
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "eagerLoadDepth descends into a level that loaded nothing", "[DataMapper][With]")
+{
+    // Owners with a region but no children at all. Depth 2 therefore resolves one relation that
+    // yields targets and one that yields none, and then recurses into both — the descent has to cope
+    // with being handed an empty batch rather than assuming every level produced rows.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 0, 0);
+
+    auto region = EagerRegion { .label = "north" };
+    dm.Create(region);
+    for (auto const index: { 0, 1 })
+    {
+        auto owner = EagerOwner { .name = SqlAnsiString<32> { std::format("childless-{}", index) } };
+        owner.region = region;
+        dm.Create(owner);
+    }
+
+    auto owners = dm.Query<EagerOwner, DataMapperOptions { .eagerLoadDepth = 2 }>().All();
+
+    REQUIRE(owners.size() == 2);
+    for (auto& owner: owners)
+    {
+        CHECK(owner.children.Count() == 0);
+        CHECK(owner.region.Record().label.Value().ToStringView() == "north");
+    }
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "A named path stops when its first level loaded nothing", "[DataMapper][With]")
+{
+    // Owners that have no children at all, asked for the two-level path owners -> children ->
+    // category. The first level resolves to an empty set of targets, so the second must not run:
+    // continuing would query the category table with an IN predicate built from nothing.
+    auto dm = DataMapper {};
+    MakeOwnersWithChildren(dm, 0, 0);
+
+    auto region = EagerRegion { .label = "north" };
+    dm.Create(region);
+    for (auto const index: { 0, 1 })
+    {
+        auto owner = EagerOwner { .name = SqlAnsiString<32> { std::format("childless-path-{}", index) } };
+        owner.region = region;
+        dm.Create(owner);
+    }
+
+    auto counter = ScopedStatementCounter {};
+    auto owners = dm.Query<EagerOwner>().With<Member(EagerOwner::children), Member(EagerChild::category)>().All();
+
+    REQUIRE(owners.size() == 2);
+    // The owner SELECT and the children SELECT — and nothing for the third level.
+    CHECK(counter.Count() == 2);
+    for (auto& owner: owners)
+        CHECK(owner.children.Count() == 0);
+}
diff --git a/src/tests/DataMapper/RelationTests.cpp b/src/tests/DataMapper/RelationTests.cpp
index 95585ab44..3a2693325 100644
--- a/src/tests/DataMapper/RelationTests.cpp
+++ b/src/tests/DataMapper/RelationTests.cpp
@@ -546,7 +546,8 @@ TEST_CASE_METHOD(SqlTestFixture, "BelongsTo", "[DataMapper][relations]")
                                     "user_id" GUID,
                                     PRIMARY KEY ("id"),
                                     CONSTRAINT "FK_Email_user_id" FOREIGN KEY ("user_id") REFERENCES "User"("id")
-                                    );)"));
+                                    );
+                                    CREATE INDEX "Email_user_id_index" ON "Email" ("user_id");)"));
     }
 }
 
@@ -1156,7 +1157,9 @@ TEST_CASE_METHOD(SqlTestFixture, "HasManyThrough", "[DataMapper][relations]")
                                     PRIMARY KEY ("id"),
                                     CONSTRAINT "FK_Appointment_physician_id" FOREIGN KEY ("physician_id") REFERENCES "Physician"("id"),
                                     CONSTRAINT "FK_Appointment_patient_id" FOREIGN KEY ("patient_id") REFERENCES "Patient"("id")
-                                    );)"));
+                                    );
+                                    CREATE INDEX "Appointment_physician_id_index" ON "Appointment" ("physician_id");
+                                    CREATE INDEX "Appointment_patient_id_index" ON "Appointment" ("patient_id");)"));
     }
 }
 
@@ -1366,7 +1369,8 @@ TEST_CASE_METHOD(SqlTestFixture, "Table with aliased column names", "[DataMapper
                                     "id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
                                     "record_id" BIGINT,
                                     CONSTRAINT "FK_BelongsToAliasedRecord_record_id" FOREIGN KEY ("record_id") REFERENCES "TheAliasedRecord"("pk")
-                                    );)"));
+                                    );
+                                    CREATE INDEX "BelongsToAliasedRecord_record_id_index" ON "BelongsToAliasedRecord" ("record_id");)"));
     }
 
     SECTION("All")
diff --git a/src/tests/SqlStatementDbTests.cpp b/src/tests/SqlStatementDbTests.cpp
index 9923f8b3b..728997695 100644
--- a/src/tests/SqlStatementDbTests.cpp
+++ b/src/tests/SqlStatementDbTests.cpp
@@ -7,6 +7,7 @@
 #include <catch2/catch_test_macros.hpp>
 
 #include <stdexcept>
+#include <utility>
 #include <vector>
 
 using namespace Lightweight;
@@ -205,3 +206,189 @@ TEST_CASE_METHOD(SqlTestFixture, "SqlResultCursor::BindOutputColumns reads multi
     CHECK(lastName == "Smith");
     CHECK(salary == 50'000);
 }
+
+// ================================================================================================
+// Prepare() reusing the statement already on the handle
+// ================================================================================================
+
+TEST_CASE_METHOD(SqlTestFixture, "Repeated Prepare of the same query keeps executing correctly", "[SqlStatement]")
+{
+    // Preparing byte-identical SQL again reuses the prepared statement instead of re-issuing
+    // SQLPrepareW. What must not change is the observable behaviour: the parameter count, the
+    // bindings, and the rows that come back have to be the same on every round.
+    auto stmt = SqlStatement {};
+    CreateEmployeesTable(stmt);
+    FillEmployeesTable(stmt);
+
+    auto const* const query = R"(SELECT "FirstName" FROM "Employees" WHERE "Salary" > ? ORDER BY "EmployeeID")";
+
+    for (auto const [threshold, expectedRows]: { std::pair { 45'000, 3 }, { 55'000, 2 }, { 65'000, 1 } })
+    {
+        stmt.Prepare(query);
+        auto cursor = stmt.Execute(threshold);
+
+        int rows = 0;
+        while (cursor.FetchRow())
+        {
+            CHECK(!cursor.GetColumn<std::string>(1).empty());
+            ++rows;
+        }
+        CHECK(rows == expectedRows);
+    }
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "Prepare of the same query survives a schema change underneath", "[SqlStatement]")
+{
+    // The reuse above is what makes this case interesting: the second Prepare() does not re-issue
+    // SQLPrepareW, so the statement the server holds may have been planned against the *old* table.
+    // PostgreSQL rejects that plan (SQLSTATE 0A000), which the statement recovers from by preparing
+    // once more - the caller must see rows, not an exception.
+    auto stmt = SqlStatement {};
+    CreateEmployeesTable(stmt);
+    FillEmployeesTable(stmt);
+
+    auto const* const query = R"(SELECT "FirstName" FROM "Employees" ORDER BY "EmployeeID")";
+
+    stmt.Prepare(query);
+    {
+        auto cursor = stmt.Execute();
+        CHECK(cursor.FetchRow());
+    }
+
+    // Rebuild the table with an extra column, through a different statement handle.
+    auto other = SqlStatement { stmt.Connection() };
+    (void) other.ExecuteDirect(R"(DROP TABLE "Employees")");
+    CreateEmployeesTable(other);
+    (void) other.ExecuteDirect(R"(ALTER TABLE "Employees" ADD "Nickname" VARCHAR(50))");
+    FillEmployeesTable(other);
+
+    stmt.Prepare(query);
+    auto cursor = stmt.Execute();
+
+    int rows = 0;
+    while (cursor.FetchRow())
+        ++rows;
+    CHECK(rows == 3);
+}
+
+TEST_CASE_METHOD(SqlTestFixture, "Prepare reuse returns correct rows after the table is recreated", "[SqlStatement]")
+{
+    using namespace Lightweight::SqlColumnTypeDefinitions;
+
+    auto connection = SqlConnection {};
+    auto stmt = SqlStatement { connection };
+
+    auto const createTable = [](SqlMigrationQueryBuilder& migration) {
+        migration.DropTableIfExists("stale_plan");
+        migration.CreateTable("stale_plan").PrimaryKey("id", Integer {}).RequiredColumn("value", Integer {});
+    };
+
+    stmt.MigrateDirect(createTable);
+    stmt.Prepare(R"(INSERT INTO "stale_plan" ("id", "value") VALUES (?, ?))");
+    std::ignore = stmt.Execute(1, 10);
+
+    auto const* const selectQuery = R"(SELECT "value" FROM "stale_plan" WHERE "id" = ?)";
+    stmt.Prepare(selectQuery);
+    {
+        auto cursor = stmt.Execute(1);
+        REQUIRE(cursor.FetchRow());
+        CHECK(cursor.GetColumn<int>(1) == 10);
+    }
+
+    // Recreate the table behind the prepared handle, through a second statement so that `stmt` keeps
+    // its prepared query and takes the reuse path below. MS SQL Server compiles a plan against the
+    // object *ids* it saw, so dropping and recreating the table invalidates the plan while the SQL
+    // text still resolves — 42S02; PostgreSQL reports the same condition as 0A000 / 26000.
+    {
+        auto ddl = SqlStatement { connection };
+        ddl.MigrateDirect(createTable);
+        ddl.Prepare(R"(INSERT INTO "stale_plan" ("id", "value") VALUES (?, ?))");
+        std::ignore = ddl.Execute(1, 99);
+    }
+
+    // Byte-identical text, so Prepare() reuses the handle rather than re-issuing SQLPrepare, and the
+    // execute must return the *new* table's rows.
+    //
+    // This is the scenario RetryStalePreparedStatement() exists for, but none of the three backends
+    // in the test matrix actually rejects the reused handle here (verified: the "Re-preparing
+    // statement" warning never fires) — MS SQL Server's Driver 18 defers preparation to execution
+    // time, so there is no server-side plan to go stale. The recovery arm therefore stays uncovered;
+    // reaching it needs a fault-injection seam rather than a real driver.
+    stmt.Prepare(selectQuery);
+    auto cursor = stmt.Execute(1);
+    REQUIRE(cursor.FetchRow());
+    CHECK(cursor.GetColumn<int>(1) == 99);
+}
+
+// ================================================================================================
+// Prepare reuse: recovering a handle the server no longer knows about
+// ================================================================================================
+
+TEST_CASE_METHOD(SqlTestFixture, "Prepare reuse recovers when the server forgot the statement", "[SqlStatement]")
+{
+    using namespace Lightweight::SqlColumnTypeDefinitions;
+
+    auto connection = SqlConnection {};
+    auto stmt = SqlStatement { connection };
+
+    // `DISCARD ALL` is the one lever in the test matrix that reliably invalidates a *reused* prepared
+    // handle: it drops every server-side prepared statement of the session, so the next execute of a
+    // handle Prepare() reused reports 26000 and RetryStalePreparedStatement has to re-prepare and run
+    // it again. Recreating the table behind the handle does not achieve this on any of the three
+    // backends (see the recreated-table test above), and `DISCARD ALL` is PostgreSQL-specific SQL
+    // besides - so this is genuinely a one-backend test rather than a dodged failure.
+    UNSUPPORTED_DATABASE(stmt, SqlServerType::SQLITE);
+    UNSUPPORTED_DATABASE(stmt, SqlServerType::MICROSOFT_SQL);
+    UNSUPPORTED_DATABASE(stmt, SqlServerType::MYSQL);
+    UNSUPPORTED_DATABASE(stmt, SqlServerType::UNKNOWN);
+
+    stmt.MigrateDirect([](SqlMigrationQueryBuilder& migration) {
+        migration.DropTableIfExists("forgotten_plan");
+        migration.CreateTable("forgotten_plan").PrimaryKey("id", Integer {}).RequiredColumn("value", Integer {});
+    });
+    std::ignore = stmt.ExecuteDirect(R"(INSERT INTO "forgotten_plan" ("id", "value") VALUES (1, 10))");
+
+    auto const* const selectQuery = R"(SELECT "value" FROM "forgotten_plan" WHERE "id" = ?)";
+
+    auto const discardServerSideStatements = [&connection] {
+        auto other = SqlStatement { connection };
+        std::ignore = other.ExecuteDirect("DISCARD ALL");
+    };
+
+    // Both execute paths carry their own retry call site, so both are driven through the recovery.
+    SECTION("typed Execute(Args...)")
+    {
+        stmt.Prepare(selectQuery);
+        {
+            auto cursor = stmt.Execute(1);
+            REQUIRE(cursor.FetchRow());
+            CHECK(cursor.GetColumn<int>(1) == 10);
+        }
+
+        discardServerSideStatements();
+
+        // Byte-identical text, so Prepare() reuses the handle instead of re-issuing SQLPrepare - which
+        // is what leaves the execute below facing a statement the server has forgotten.
+        stmt.Prepare(selectQuery);
+        auto cursor = stmt.Execute(1);
+        REQUIRE(cursor.FetchRow());
+        CHECK(cursor.GetColumn<int>(1) == 10);
+    }
+
+    SECTION("ExecuteWithVariants")
+    {
+        stmt.Prepare(selectQuery);
+        {
+            auto cursor = stmt.ExecuteWithVariants({ SqlVariant { 1 } });
+            REQUIRE(cursor.FetchRow());
+            CHECK(cursor.GetColumn<int>(1) == 10);
+        }
+
+        discardServerSideStatements();
+
+        stmt.Prepare(selectQuery);
+        auto cursor = stmt.ExecuteWithVariants({ SqlVariant { 1 } });
+        REQUIRE(cursor.FetchRow());
+        CHECK(cursor.GetColumn<int>(1) == 10);
+    }
+}