diff --git a/lib/Doctrine/ORM/Persisters/Entity/AbstractEntityInheritancePersister.php b/lib/Doctrine/ORM/Persisters/Entity/AbstractEntityInheritancePersister.php
index edb39afc6..6b5e5c4f1 100644
--- a/lib/Doctrine/ORM/Persisters/Entity/AbstractEntityInheritancePersister.php
+++ b/lib/Doctrine/ORM/Persisters/Entity/AbstractEntityInheritancePersister.php
@@ -66,7 +66,7 @@ abstract class AbstractEntityInheritancePersister extends BasicEntityPersister
         $sql         = $this->getSQLTableAlias($class->name, $tableAlias) . '.'
                             . $this->quoteStrategy->getColumnName($field, $class, $this->platform);
 
-        $this->rsm->addFieldResult($alias, $columnAlias, $field, $class->name);
+        $this->currentPersisterContext->rsm->addFieldResult($alias, $columnAlias, $field, $class->name);
 
         if (isset($class->fieldMappings[$field]['requireSQLConversion'])) {
             $type   = Type::getType($class->getTypeOfField($field));
@@ -88,7 +88,7 @@ abstract class AbstractEntityInheritancePersister extends BasicEntityPersister
     {
         $columnAlias = $this->getSQLColumnAlias($joinColumnName);
 
-        $this->rsm->addMetaResult('r', $columnAlias, $joinColumnName, false, $type);
+        $this->currentPersisterContext->rsm->addMetaResult('r', $columnAlias, $joinColumnName, false, $type);
 
         return $tableAlias . '.' . $joinColumnName . ' AS ' . $columnAlias;
     }
diff --git a/lib/Doctrine/ORM/Persisters/Entity/BasicEntityPersister.php b/lib/Doctrine/ORM/Persisters/Entity/BasicEntityPersister.php
index c53e817c1..a3f35f440 100644
--- a/lib/Doctrine/ORM/Persisters/Entity/BasicEntityPersister.php
+++ b/lib/Doctrine/ORM/Persisters/Entity/BasicEntityPersister.php
@@ -134,15 +134,6 @@ class BasicEntityPersister implements EntityPersister
      */
     protected $queuedInserts = array();
 
-    /**
-     * ResultSetMapping that is used for all queries. Is generated lazily once per request.
-     *
-     * TODO: Evaluate Caching in combination with the other cached SQL snippets.
-     *
-     * @var Query\ResultSetMapping
-     */
-    protected $rsm;
-
     /**
      * The map of column names to DBAL mapping types of all prepared columns used
      * when INSERTing or UPDATEing an entity.
@@ -172,36 +163,6 @@ class BasicEntityPersister implements EntityPersister
      */
     private $insertSql;
 
-    /**
-     * The SELECT column list SQL fragment used for querying entities by this persister.
-     * This SQL fragment is only generated once per request, if at all.
-     *
-     * @var string
-     */
-    protected $selectColumnListSql;
-
-    /**
-     * The JOIN SQL fragment used to eagerly load all many-to-one and one-to-one
-     * associations configured as FETCH_EAGER, as well as all inverse one-to-one associations.
-     *
-     * @var string
-     */
-    protected $selectJoinSql;
-
-    /**
-     * Counter for creating unique SQL table and column aliases.
-     *
-     * @var integer
-     */
-    protected $sqlAliasCounter = 0;
-
-    /**
-     * Map from class names (FQCN) to the corresponding generated SQL table aliases.
-     *
-     * @var array
-     */
-    protected $sqlTableAliases = array();
-
     /**
      * The quote strategy.
      *
@@ -216,6 +177,21 @@ class BasicEntityPersister implements EntityPersister
      */
     private $identifierFlattener;
 
+    /**
+     * @var CachedPersisterContext
+     */
+    protected $currentPersisterContext;
+
+    /**
+     * @var CachedPersisterContext
+     */
+    private $limitsHandlingContext;
+
+    /**
+     * @var CachedPersisterContext
+     */
+    private $noLimitsContext;
+
     /**
      * Initializes a new <tt>BasicEntityPersister</tt> that uses the given EntityManager
      * and persists instances of the class described by the given ClassMetadata descriptor.
@@ -225,12 +201,22 @@ class BasicEntityPersister implements EntityPersister
      */
     public function __construct(EntityManagerInterface $em, ClassMetadata $class)
     {
-        $this->em                  = $em;
-        $this->class               = $class;
-        $this->conn                = $em->getConnection();
-        $this->platform            = $this->conn->getDatabasePlatform();
-        $this->quoteStrategy       = $em->getConfiguration()->getQuoteStrategy();
-        $this->identifierFlattener = new IdentifierFlattener($em->getUnitOfWork(), $em->getMetadataFactory());
+        $this->em                    = $em;
+        $this->class                 = $class;
+        $this->conn                  = $em->getConnection();
+        $this->platform              = $this->conn->getDatabasePlatform();
+        $this->quoteStrategy         = $em->getConfiguration()->getQuoteStrategy();
+        $this->identifierFlattener   = new IdentifierFlattener($em->getUnitOfWork(), $em->getMetadataFactory());
+        $this->noLimitsContext       = $this->currentPersisterContext = new CachedPersisterContext(
+            $class,
+            new Query\ResultSetMapping(),
+            false
+        );
+        $this->limitsHandlingContext = new CachedPersisterContext(
+            $class,
+            new Query\ResultSetMapping(),
+            true
+        );
     }
 
     /**
@@ -246,7 +232,7 @@ class BasicEntityPersister implements EntityPersister
      */
     public function getResultSetMapping()
     {
-        return $this->rsm;
+        return $this->currentPersisterContext->rsm;
     }
 
     /**
@@ -710,6 +696,8 @@ class BasicEntityPersister implements EntityPersister
      */
     public function load(array $criteria, $entity = null, $assoc = null, array $hints = array(), $lockMode = null, $limit = null, array $orderBy = null)
     {
+        $this->switchPersisterContext(null, $limit);
+
         $sql = $this->getSelectSQL($criteria, $assoc, $lockMode, $limit, null, $orderBy);
         list($params, $types) = $this->expandParameters($criteria);
         $stmt = $this->conn->executeQuery($sql, $params, $types);
@@ -719,8 +707,8 @@ class BasicEntityPersister implements EntityPersister
             $hints[Query::HINT_REFRESH_ENTITY]  = $entity;
         }
 
-        $hydrator = $this->em->newHydrator($this->selectJoinSql ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
-        $entities = $hydrator->hydrateAll($stmt, $this->rsm, $hints);
+        $hydrator = $this->em->newHydrator($this->currentPersisterContext->selectJoinSql ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
+        $entities = $hydrator->hydrateAll($stmt, $this->currentPersisterContext->rsm, $hints);
 
         return $entities ? $entities[0] : null;
     }
@@ -809,7 +797,7 @@ class BasicEntityPersister implements EntityPersister
         $stmt = $this->conn->executeQuery($sql, $params, $types);
 
         $hydrator = $this->em->newHydrator(Query::HYDRATE_OBJECT);
-        $hydrator->hydrateAll($stmt, $this->rsm, array(Query::HINT_REFRESH => true));
+        $hydrator->hydrateAll($stmt, $this->currentPersisterContext->rsm, array(Query::HINT_REFRESH => true));
     }
 
     /**
@@ -839,9 +827,9 @@ class BasicEntityPersister implements EntityPersister
         list($params, $types) = $this->expandCriteriaParameters($criteria);
 
         $stmt       = $this->conn->executeQuery($query, $params, $types);
-        $hydrator   = $this->em->newHydrator(($this->selectJoinSql) ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
+        $hydrator   = $this->em->newHydrator(($this->currentPersisterContext->selectJoinSql) ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
 
-        return $hydrator->hydrateAll($stmt, $this->rsm, array(UnitOfWork::HINT_DEFEREAGERLOAD => true));
+        return $hydrator->hydrateAll($stmt, $this->currentPersisterContext->rsm, array(UnitOfWork::HINT_DEFEREAGERLOAD => true));
     }
 
     /**
@@ -880,13 +868,15 @@ class BasicEntityPersister implements EntityPersister
      */
     public function loadAll(array $criteria = array(), array $orderBy = null, $limit = null, $offset = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $sql = $this->getSelectSQL($criteria, null, null, $limit, $offset, $orderBy);
         list($params, $types) = $this->expandParameters($criteria);
         $stmt = $this->conn->executeQuery($sql, $params, $types);
 
-        $hydrator = $this->em->newHydrator(($this->selectJoinSql) ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
+        $hydrator = $this->em->newHydrator(($this->currentPersisterContext->selectJoinSql) ? Query::HYDRATE_OBJECT : Query::HYDRATE_SIMPLEOBJECT);
 
-        return $hydrator->hydrateAll($stmt, $this->rsm, array(UnitOfWork::HINT_DEFEREAGERLOAD => true));
+        return $hydrator->hydrateAll($stmt, $this->currentPersisterContext->rsm, array(UnitOfWork::HINT_DEFEREAGERLOAD => true));
     }
 
     /**
@@ -894,6 +884,8 @@ class BasicEntityPersister implements EntityPersister
      */
     public function getManyToManyCollection(array $assoc, $sourceEntity, $offset = null, $limit = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $stmt = $this->getManyToManyStatement($assoc, $sourceEntity, $offset, $limit);
 
         return $this->loadArrayFromStatement($assoc, $stmt);
@@ -909,11 +901,11 @@ class BasicEntityPersister implements EntityPersister
      */
     private function loadArrayFromStatement($assoc, $stmt)
     {
-        $rsm    = $this->rsm;
+        $rsm    = $this->currentPersisterContext->rsm;
         $hints  = array(UnitOfWork::HINT_DEFEREAGERLOAD => true);
 
         if (isset($assoc['indexBy'])) {
-            $rsm = clone ($this->rsm); // this is necessary because the "default rsm" should be changed.
+            $rsm = clone ($this->currentPersisterContext->rsm); // this is necessary because the "default rsm" should be changed.
             $rsm->addIndexBy('r', $assoc['indexBy']);
         }
 
@@ -931,14 +923,14 @@ class BasicEntityPersister implements EntityPersister
      */
     private function loadCollectionFromStatement($assoc, $stmt, $coll)
     {
-        $rsm   = $this->rsm;
+        $rsm   = $this->currentPersisterContext->rsm;
         $hints = array(
             UnitOfWork::HINT_DEFEREAGERLOAD => true,
             'collection' => $coll
         );
 
         if (isset($assoc['indexBy'])) {
-            $rsm = clone ($this->rsm); // this is necessary because the "default rsm" should be changed.
+            $rsm = clone ($this->currentPersisterContext->rsm); // this is necessary because the "default rsm" should be changed.
             $rsm->addIndexBy('r', $assoc['indexBy']);
         }
 
@@ -967,6 +959,8 @@ class BasicEntityPersister implements EntityPersister
      */
     private function getManyToManyStatement(array $assoc, $sourceEntity, $offset = null, $limit = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $sourceClass    = $this->em->getClassMetadata($assoc['sourceEntity']);
         $class          = $sourceClass;
         $association    = $assoc;
@@ -1032,6 +1026,8 @@ class BasicEntityPersister implements EntityPersister
      */
     public function getSelectSQL($criteria, $assoc = null, $lockMode = null, $limit = null, $offset = null, array $orderBy = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $lockSql    = '';
         $joinSql    = '';
         $orderBySql = '';
@@ -1075,7 +1071,7 @@ class BasicEntityPersister implements EntityPersister
 
         $select = 'SELECT ' . $columnList;
         $from   = ' FROM ' . $tableName . ' '. $tableAlias;
-        $join   = $this->selectJoinSql . $joinSql;
+        $join   = $this->currentPersisterContext->selectJoinSql . $joinSql;
         $where  = ($conditionSql ? ' WHERE ' . $conditionSql : '');
         $lock   = $this->platform->appendLockHint($from, $lockMode);
         $query  = $select
@@ -1184,20 +1180,19 @@ class BasicEntityPersister implements EntityPersister
      */
     protected function getSelectColumnsSQL()
     {
-        if ($this->selectColumnListSql !== null) {
-            return $this->selectColumnListSql;
+        if ($this->currentPersisterContext->selectColumnListSql !== null) {
+            return $this->currentPersisterContext->selectColumnListSql;
         }
 
         $columnList = array();
-        $this->rsm  = new Query\ResultSetMapping();
-        $this->rsm->addEntityResult($this->class->name, 'r'); // r for root
+        $this->currentPersisterContext->rsm->addEntityResult($this->class->name, 'r'); // r for root
 
         // Add regular columns to select list
         foreach ($this->class->fieldNames as $field) {
             $columnList[] = $this->getSelectColumnSQL($field, $this->class);
         }
 
-        $this->selectJoinSql    = '';
+        $this->currentPersisterContext->selectJoinSql    = '';
         $eagerAliasCounter      = 0;
 
         foreach ($this->class->associationMappings as $assocField => $assoc) {
@@ -1214,6 +1209,10 @@ class BasicEntityPersister implements EntityPersister
                 continue;
             }
 
+            if ((($assoc['type'] & ClassMetadata::TO_MANY) > 0) && $this->currentPersisterContext->handlesLimits) {
+                continue;
+            }
+
             $eagerEntity = $this->em->getClassMetadata($assoc['targetEntity']);
 
             if ($eagerEntity->inheritanceType != ClassMetadata::INHERITANCE_TYPE_NONE) {
@@ -1221,7 +1220,7 @@ class BasicEntityPersister implements EntityPersister
             }
 
             $assocAlias = 'e' . ($eagerAliasCounter++);
-            $this->rsm->addJoinedEntityResult($assoc['targetEntity'], $assocAlias, 'r', $assocField);
+            $this->currentPersisterContext->rsm->addJoinedEntityResult($assoc['targetEntity'], $assocAlias, 'r', $assocField);
 
             foreach ($eagerEntity->fieldNames as $field) {
                 $columnList[] = $this->getSelectColumnSQL($field, $eagerEntity, $assocAlias);
@@ -1241,7 +1240,7 @@ class BasicEntityPersister implements EntityPersister
             $joinCondition  = array();
 
             if (isset($assoc['indexBy'])) {
-                $this->rsm->addIndexBy($assocAlias, $assoc['indexBy']);
+                $this->currentPersisterContext->rsm->addIndexBy($assocAlias, $assoc['indexBy']);
             }
 
             if ( ! $assoc['isOwningSide']) {
@@ -1254,7 +1253,7 @@ class BasicEntityPersister implements EntityPersister
 
             if ($assoc['isOwningSide']) {
                 $tableAlias           = $this->getSQLTableAlias($association['targetEntity'], $assocAlias);
-                $this->selectJoinSql .= ' ' . $this->getJoinSQLForJoinColumns($association['joinColumns']);
+                $this->currentPersisterContext->selectJoinSql .= ' ' . $this->getJoinSQLForJoinColumns($association['joinColumns']);
 
                 foreach ($association['joinColumns'] as $joinColumn) {
                     $sourceCol       = $this->quoteStrategy->getJoinColumnName($joinColumn, $this->class, $this->platform);
@@ -1270,7 +1269,7 @@ class BasicEntityPersister implements EntityPersister
 
             } else {
 
-                $this->selectJoinSql .= ' LEFT JOIN';
+                $this->currentPersisterContext->selectJoinSql .= ' LEFT JOIN';
 
                 foreach ($association['joinColumns'] as $joinColumn) {
                     $sourceCol       = $this->quoteStrategy->getJoinColumnName($joinColumn, $this->class, $this->platform);
@@ -1281,13 +1280,13 @@ class BasicEntityPersister implements EntityPersister
                 }
             }
 
-            $this->selectJoinSql .= ' ' . $joinTableName . ' ' . $joinTableAlias . ' ON ';
-            $this->selectJoinSql .= implode(' AND ', $joinCondition);
+            $this->currentPersisterContext->selectJoinSql .= ' ' . $joinTableName . ' ' . $joinTableAlias . ' ON ';
+            $this->currentPersisterContext->selectJoinSql .= implode(' AND ', $joinCondition);
         }
 
-        $this->selectColumnListSql = implode(', ', $columnList);
+        $this->currentPersisterContext->selectColumnListSql = implode(', ', $columnList);
 
-        return $this->selectColumnListSql;
+        return $this->currentPersisterContext->selectColumnListSql;
     }
 
     /**
@@ -1318,7 +1317,7 @@ class BasicEntityPersister implements EntityPersister
                                 . '.' . $quotedColumn . ' AS ' . $resultColumnName;
             $type             = PersisterHelper::getTypeOfColumn($joinColumn['referencedColumnName'], $targetClass, $this->em);
 
-            $this->rsm->addMetaResult($alias, $resultColumnName, $quotedColumn, $isIdentifier, $type);
+            $this->currentPersisterContext->rsm->addMetaResult($alias, $resultColumnName, $quotedColumn, $isIdentifier, $type);
         }
 
         return implode(', ', $columnList);
@@ -1456,7 +1455,7 @@ class BasicEntityPersister implements EntityPersister
         $sql            = $tableAlias . '.' . $columnName;
         $columnAlias    = $this->getSQLColumnAlias($class->columnNames[$field]);
 
-        $this->rsm->addFieldResult($alias, $columnAlias, $field);
+        $this->currentPersisterContext->rsm->addFieldResult($alias, $columnAlias, $field);
 
         if (isset($class->fieldMappings[$field]['requireSQLConversion'])) {
             $type   = Type::getType($class->getTypeOfField($field));
@@ -1482,13 +1481,13 @@ class BasicEntityPersister implements EntityPersister
             $className .= '#' . $assocName;
         }
 
-        if (isset($this->sqlTableAliases[$className])) {
-            return $this->sqlTableAliases[$className];
+        if (isset($this->currentPersisterContext->sqlTableAliases[$className])) {
+            return $this->currentPersisterContext->sqlTableAliases[$className];
         }
 
-        $tableAlias = 't' . $this->sqlAliasCounter++;
+        $tableAlias = 't' . $this->currentPersisterContext->sqlAliasCounter++;
 
-        $this->sqlTableAliases[$className] = $tableAlias;
+        $this->currentPersisterContext->sqlTableAliases[$className] = $tableAlias;
 
         return $tableAlias;
     }
@@ -1690,6 +1689,8 @@ class BasicEntityPersister implements EntityPersister
      */
     public function getOneToManyCollection(array $assoc, $sourceEntity, $offset = null, $limit = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $stmt = $this->getOneToManyStatement($assoc, $sourceEntity, $offset, $limit);
 
         return $this->loadArrayFromStatement($assoc, $stmt);
@@ -1717,6 +1718,8 @@ class BasicEntityPersister implements EntityPersister
      */
     private function getOneToManyStatement(array $assoc, $sourceEntity, $offset = null, $limit = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $criteria    = array();
         $parameters  = array();
         $owningAssoc = $this->class->associationMappings[$assoc['mappedBy']];
@@ -1930,7 +1933,7 @@ class BasicEntityPersister implements EntityPersister
      */
     public function getSQLColumnAlias($columnName)
     {
-        return $this->quoteStrategy->getColumnAlias($columnName, $this->sqlAliasCounter++, $this->platform);
+        return $this->quoteStrategy->getColumnAlias($columnName, $this->currentPersisterContext->sqlAliasCounter++, $this->platform);
     }
 
     /**
@@ -1954,4 +1957,23 @@ class BasicEntityPersister implements EntityPersister
         $sql = implode(' AND ', $filterClauses);
         return $sql ? "(" . $sql . ")" : ""; // Wrap again to avoid "X or Y and FilterConditionSQL"
     }
+
+    /**
+     * Switches persister context according to current query offset/limits
+     *
+     * This is due to the fact that to-many associations cannot be fetch-joined when a limit is involved
+     *
+     * @param null|int $offset
+     * @param null|int $limit
+     */
+    protected function switchPersisterContext($offset, $limit)
+    {
+        if (null === $offset && null === $limit) {
+            $this->currentPersisterContext = $this->noLimitsContext;
+
+            return;
+        }
+
+        $this->currentPersisterContext = $this->limitsHandlingContext;
+    }
 }
diff --git a/lib/Doctrine/ORM/Persisters/Entity/CachedPersisterContext.php b/lib/Doctrine/ORM/Persisters/Entity/CachedPersisterContext.php
new file mode 100644
index 000000000..81fde938b
--- /dev/null
+++ b/lib/Doctrine/ORM/Persisters/Entity/CachedPersisterContext.php
@@ -0,0 +1,102 @@
+<?php
+/*
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ * This software consists of voluntary contributions made by many individuals
+ * and is licensed under the MIT license. For more information, see
+ * <http://www.doctrine-project.org>.
+ */
+
+namespace Doctrine\ORM\Persisters\Entity;
+use Doctrine\Common\Persistence\Mapping\ClassMetadata;
+use Doctrine\ORM\Query\ResultSetMapping;
+
+/**
+ * A swappable persister context to use as a container for the current
+ * generated query/resultSetMapping/type binding information.
+ *
+ * This class is a utility class to be used only by the persister API
+ *
+ * This object is highly mutable due to performance reasons. Same reasoning
+ * behind its properties being public.
+ *
+ * @author Marco Pivetta <ocramius@gmail.com>
+ */
+class CachedPersisterContext
+{
+    /**
+     * Metadata object that describes the mapping of the mapped entity class.
+     *
+     * @var \Doctrine\ORM\Mapping\ClassMetadata
+     */
+    public $class;
+
+    /**
+     * ResultSetMapping that is used for all queries. Is generated lazily once per request.
+     *
+     * @var \Doctrine\ORM\Query\ResultSetMapping
+     */
+    public $rsm;
+
+    /**
+     * The SELECT column list SQL fragment used for querying entities by this persister.
+     * This SQL fragment is only generated once per request, if at all.
+     *
+     * @var string|null
+     */
+    public $selectColumnListSql;
+
+    /**
+     * The JOIN SQL fragment used to eagerly load all many-to-one and one-to-one
+     * associations configured as FETCH_EAGER, as well as all inverse one-to-one associations.
+     *
+     * @var string
+     */
+    public $selectJoinSql;
+
+    /**
+     * Counter for creating unique SQL table and column aliases.
+     *
+     * @var integer
+     */
+    public $sqlAliasCounter = 0;
+
+    /**
+     * Map from class names (FQCN) to the corresponding generated SQL table aliases.
+     *
+     * @var array
+     */
+    public $sqlTableAliases = array();
+
+    /**
+     * Whether this persistent context is considering limit operations applied to the selection queries
+     *
+     * @var bool
+     */
+    public $handlesLimits;
+
+    /**
+     * @param ClassMetadata    $class
+     * @param ResultSetMapping $rsm
+     * @param bool             $handlesLimits
+     */
+    public function __construct(
+        ClassMetadata $class,
+        ResultSetMapping $rsm,
+        $handlesLimits
+    ) {
+        $this->class         = $class;
+        $this->rsm           = $rsm;
+        $this->handlesLimits = (bool) $handlesLimits;
+    }
+}
diff --git a/lib/Doctrine/ORM/Persisters/Entity/JoinedSubclassPersister.php b/lib/Doctrine/ORM/Persisters/Entity/JoinedSubclassPersister.php
index 019d987d5..a5bd84348 100644
--- a/lib/Doctrine/ORM/Persisters/Entity/JoinedSubclassPersister.php
+++ b/lib/Doctrine/ORM/Persisters/Entity/JoinedSubclassPersister.php
@@ -299,6 +299,8 @@ class JoinedSubclassPersister extends AbstractEntityInheritancePersister
      */
     public function getSelectSQL($criteria, $assoc = null, $lockMode = null, $limit = null, $offset = null, array $orderBy = null)
     {
+        $this->switchPersisterContext($offset, $limit);
+
         $baseTableAlias = $this->getSQLTableAlias($this->class->name);
         $joinSql        = $this->getJoinSql($baseTableAlias);
 
@@ -420,19 +422,18 @@ class JoinedSubclassPersister extends AbstractEntityInheritancePersister
     protected function getSelectColumnsSQL()
     {
         // Create the column list fragment only once
-        if ($this->selectColumnListSql !== null) {
-            return $this->selectColumnListSql;
+        if ($this->currentPersisterContext->selectColumnListSql !== null) {
+            return $this->currentPersisterContext->selectColumnListSql;
         }
 
         $columnList         = array();
-        $this->rsm          = new ResultSetMapping();
         $discrColumn        = $this->class->discriminatorColumn['name'];
         $baseTableAlias     = $this->getSQLTableAlias($this->class->name);
         $resultColumnName   = $this->platform->getSQLResultCasing($discrColumn);
 
-        $this->rsm->addEntityResult($this->class->name, 'r');
-        $this->rsm->setDiscriminatorColumn('r', $resultColumnName);
-        $this->rsm->addMetaResult('r', $resultColumnName, $discrColumn);
+        $this->currentPersisterContext->rsm->addEntityResult($this->class->name, 'r');
+        $this->currentPersisterContext->rsm->setDiscriminatorColumn('r', $resultColumnName);
+        $this->currentPersisterContext->rsm->addMetaResult('r', $resultColumnName, $discrColumn);
 
         // Add regular columns
         foreach ($this->class->fieldMappings as $fieldName => $mapping) {
@@ -523,9 +524,9 @@ class JoinedSubclassPersister extends AbstractEntityInheritancePersister
             }
         }
 
-        $this->selectColumnListSql = implode(', ', $columnList);
+        $this->currentPersisterContext->selectColumnListSql = implode(', ', $columnList);
 
-        return $this->selectColumnListSql;
+        return $this->currentPersisterContext->selectColumnListSql;
     }
 
     /**
diff --git a/lib/Doctrine/ORM/Persisters/Entity/SingleTablePersister.php b/lib/Doctrine/ORM/Persisters/Entity/SingleTablePersister.php
index 23ebdf2d2..f9e4d5573 100644
--- a/lib/Doctrine/ORM/Persisters/Entity/SingleTablePersister.php
+++ b/lib/Doctrine/ORM/Persisters/Entity/SingleTablePersister.php
@@ -48,8 +48,8 @@ class SingleTablePersister extends AbstractEntityInheritancePersister
      */
     protected function getSelectColumnsSQL()
     {
-        if ($this->selectColumnListSql !== null) {
-            return $this->selectColumnListSql;
+        if ($this->currentPersisterContext->selectColumnListSql !== null) {
+            return $this->currentPersisterContext->selectColumnListSql;
         }
 
         $columnList[] = parent::getSelectColumnsSQL();
@@ -63,8 +63,8 @@ class SingleTablePersister extends AbstractEntityInheritancePersister
 
         $resultColumnName = $this->platform->getSQLResultCasing($discrColumn);
 
-        $this->rsm->setDiscriminatorColumn('r', $resultColumnName);
-        $this->rsm->addMetaResult('r', $resultColumnName, $discrColumn);
+        $this->currentPersisterContext->rsm->setDiscriminatorColumn('r', $resultColumnName);
+        $this->currentPersisterContext->rsm->addMetaResult('r', $resultColumnName, $discrColumn);
 
         // Append subclass columns
         foreach ($this->class->subClasses as $subClassName) {
@@ -106,9 +106,9 @@ class SingleTablePersister extends AbstractEntityInheritancePersister
             }
         }
 
-        $this->selectColumnListSql = implode(', ', $columnList);
+        $this->currentPersisterContext->selectColumnListSql = implode(', ', $columnList);
 
-        return $this->selectColumnListSql;
+        return $this->currentPersisterContext->selectColumnListSql;
     }
 
     /**
diff --git a/tests/Doctrine/Tests/Models/DDC3346/DDC3346Article.php b/tests/Doctrine/Tests/Models/DDC3346/DDC3346Article.php
new file mode 100644
index 000000000..dc8a876c5
--- /dev/null
+++ b/tests/Doctrine/Tests/Models/DDC3346/DDC3346Article.php
@@ -0,0 +1,27 @@
+<?php
+
+namespace Doctrine\Tests\Models\DDC3346;
+
+/**
+ * @Entity
+ * @Table(name="ddc3346_articles")
+ */
+class DDC3346Article
+{
+    const CLASSNAME = __CLASS__;
+
+    /**
+     * @Id
+     * @Column(type="integer")
+     * @GeneratedValue(strategy="AUTO")
+     */
+    public $id;
+
+    /**
+     * @var DDC3346Author
+     *
+     * @ManyToOne(targetEntity="DDC3346Author", inversedBy="articles")
+     * @JoinColumn(name="user_id", referencedColumnName="id")
+     */
+    public $user;
+}
diff --git a/tests/Doctrine/Tests/Models/DDC3346/DDC3346Author.php b/tests/Doctrine/Tests/Models/DDC3346/DDC3346Author.php
new file mode 100644
index 000000000..5229ee280
--- /dev/null
+++ b/tests/Doctrine/Tests/Models/DDC3346/DDC3346Author.php
@@ -0,0 +1,28 @@
+<?php
+
+namespace Doctrine\Tests\Models\DDC3346;
+
+/**
+ * @Entity
+ * @Table(name="ddc3346_users")
+ */
+class DDC3346Author
+{
+    const CLASSNAME = __CLASS__;
+
+    /**
+     * @Id @Column(type="integer")
+     * @GeneratedValue
+     */
+    public $id;
+
+    /**
+     * @Column(type="string", length=255, unique=true)
+     */
+    public $username;
+
+    /**
+     * @OneToMany(targetEntity="DDC3346Article", mappedBy="user", fetch="EAGER", cascade={"detach"})
+     */
+    public $articles = array();
+}
diff --git a/tests/Doctrine/Tests/ORM/Functional/Ticket/DDC3346Test.php b/tests/Doctrine/Tests/ORM/Functional/Ticket/DDC3346Test.php
new file mode 100644
index 000000000..ba286da24
--- /dev/null
+++ b/tests/Doctrine/Tests/ORM/Functional/Ticket/DDC3346Test.php
@@ -0,0 +1,77 @@
+<?php
+
+namespace Doctrine\Tests\ORM\Functional\Ticket;
+
+use Doctrine\Tests\Models\DDC3346\DDC3346Article;
+use Doctrine\Tests\Models\DDC3346\DDC3346Author;
+
+/**
+ * @group DDC-3346
+ */
+class DDC3346Test extends \Doctrine\Tests\OrmFunctionalTestCase
+{
+    public function setUp()
+    {
+        $this->useModelSet('ddc3346');
+
+        parent::setUp();
+
+        $this->loadAuthorFixture();
+    }
+
+    public function testFindOneWithEagerFetchWillNotHydrateLimitedCollection()
+    {
+        /* @var DDC3346Author $author */
+        $author = $this->_em->getRepository(DDC3346Author::CLASSNAME)->findOneBy(
+            array('username' => 'bwoogy')
+        );
+
+        $this->assertCount(2, $author->articles);
+    }
+
+    public function testFindLimitedWithEagerFetchWillNotHydrateLimitedCollection()
+    {
+        /* @var DDC3346Author[] $authors */
+        $authors = $this->_em->getRepository(DDC3346Author::CLASSNAME)->findBy(
+            array('username' => 'bwoogy'),
+            null,
+            1
+        );
+
+        $this->assertCount(1, $authors);
+        $this->assertCount(2, $authors[0]->articles);
+    }
+
+    public function testFindWithEagerFetchAndOffsetWillNotHydrateLimitedCollection()
+    {
+        /* @var DDC3346Author[] $authors */
+        $authors = $this->_em->getRepository(DDC3346Author::CLASSNAME)->findBy(
+            array('username' => 'bwoogy'),
+            null,
+            null,
+            0 // using an explicitly defined offset
+        );
+
+        $this->assertCount(1, $authors);
+        $this->assertCount(2, $authors[0]->articles);
+    }
+
+    private function loadAuthorFixture()
+    {
+        $user     = new DDC3346Author();
+        $article1 = new DDC3346Article();
+        $article2 = new DDC3346Article();
+
+        $user->username   = 'bwoogy';
+        $article1->user   = $user;
+        $article2->user   = $user;
+        $user->articles[] = $article1;
+        $user->articles[] = $article2;
+
+        $this->_em->persist($user);
+        $this->_em->persist($article1);
+        $this->_em->persist($article2);
+        $this->_em->flush();
+        $this->_em->clear();
+    }
+}
diff --git a/tests/Doctrine/Tests/OrmFunctionalTestCase.php b/tests/Doctrine/Tests/OrmFunctionalTestCase.php
index cfced6d61..f73ac54fc 100644
--- a/tests/Doctrine/Tests/OrmFunctionalTestCase.php
+++ b/tests/Doctrine/Tests/OrmFunctionalTestCase.php
@@ -193,6 +193,10 @@ abstract class OrmFunctionalTestCase extends OrmTestCase
             'Doctrine\Tests\Models\DDC2504\DDC2504ChildClass',
             'Doctrine\Tests\Models\DDC2504\DDC2504OtherClass',
         ),
+        'ddc3346' => array(
+            'Doctrine\Tests\Models\DDC3346\DDC3346Author',
+            'Doctrine\Tests\Models\DDC3346\DDC3346Article',
+        ),
         'quote' => array(
             'Doctrine\Tests\Models\Quote\Address',
             'Doctrine\Tests\Models\Quote\Group',
@@ -399,6 +403,11 @@ abstract class OrmFunctionalTestCase extends OrmTestCase
             $conn->executeUpdate('DELETE FROM cache_country');
         }
 
+        if (isset($this->_usedModelSets['ddc3346'])) {
+            $conn->executeUpdate('DELETE FROM ddc3346_articles');
+            $conn->executeUpdate('DELETE FROM ddc3346_users');
+        }
+
         if (isset($this->_usedModelSets['quote'])) {
             $conn->executeUpdate('DELETE FROM ' . $platform->quoteIdentifier("quote-address"));
             $conn->executeUpdate('DELETE FROM ' . $platform->quoteIdentifier("quote-group"));