Initial implementation of pagination for older Sqlserver

Old SQLServer does not support anything like LIMIT/OFFSET. Instead we
have to use a subquery workaround.

src/Database/Dialect/SqlserverDialectTrait.php

@@ -16,8 +16,12 @@
 
 use Cake\Database\Dialect\TupleComparisonTranslatorTrait;
 use Cake\Database\Expression\FunctionExpression;
+use Cake\Database\Expression\OrderByExpression;
+use Cake\Database\Expression\UnaryExpression;
+use Cake\Database\Query;
 use Cake\Database\SqlDialectTrait;
 use Cake\Database\SqlserverCompiler;
+use PDO;
 
 /**
  * Contains functions that encapsulates the SQL dialect used by SQLServer,
@@ -60,9 +64,60 @@ protected function _selectQueryTranslator($query) {
 			$query->order($query->newExpr()->add('SELECT NULL'));
 		}
 
+		if ($this->_version() < 11 && $offset) {
+			return $this->_pagingSubquery($query, $limit, $offset);
+		}
+
 		return $query;
 	}
 
+/**
+ * Get the version of SQLserver we are connected to.
+ *
+ * @return int
+ */
+	public function _version() {
+		return $this->_connection->getAttribute(PDO::ATTR_SERVER_VERSION);
+	}
+
+/**
+ * Generate a paging subquery for older versions of SQLserver.
+ *
+ * Prior to SQLServer 2012 there was no equivalent to LIMIT OFFSET, so a subquery must
+ * be used.
+ *
+ * @param \Cake\Database\Query $query The query to wrap in a subquery.
+ * @param int $limit The number of rows to fetch.
+ * @param int $offset The number of rows to offset.
+ * @return \Cake\Database\Query Modified query object.
+ */
+	protected function _pagingSubquery($query, $limit, $offset) {
+		$field = '_cake_paging_._cake_page_rownum_';
+
+		$order = $query->clause('order') ?: new OrderByExpression('NULL');
+		$query->select([
+				'_cake_page_rownum_' => new UnaryExpression($order, [], 'ROW_NUMBER() OVER')
+			])->limit(null)
+			->offset(null)
+			->order([], true);
+
+		$outer = new Query($query->connection());
+		$outer->select('*')
+			->from(['_cake_paging_' => $query]);
+		if ($offset) {
+			$outer->where(["$field >" => $offset]);
+		}
+		if ($limit) {
+			$outer->where(["$field <=" => (int)$offset + (int)$limit]);
+		}
+		return $outer->decorateResults(function ($row) {
+			if (isset($row['_cake_page_rownum_'])) {
+				unset($row['_cake_page_rownum_']);
+			}
+			return $row;
+		});
+	}
+
 /**
  * Returns a dictionary of expressions to be transformed when compiling a Query
  * to SQL. Array keys are method names to be called in this class

tests/TestCase/Database/Driver/SqlserverTest.php

@@ -91,16 +91,21 @@ public function testConnectionConfigCustom() {
 	}
 
 /**
- * Test select with limit only
+ * Test select with limit only and SQLServer2012+
  *
  * @return void
  */
-	public function testSelectLimit() {
+	public function testSelectLimitVersion12() {
 		$driver = $this->getMock(
 			'Cake\Database\Driver\Sqlserver',
-			['_connect', 'connection'],
+			['_connect', 'connection', '_version'],
 			[['dsn' => 'foo']]
 		);
+		$driver
+			->expects($this->any())
+			->method('_version')
+			->will($this->returnValue(12));
+
 		$connection = $this->getMock(
 			'\Cake\Database\Connection',
 			['connect', 'driver'],
@@ -139,4 +144,68 @@ public function testSelectLimit() {
 		$this->assertEquals('SELECT TOP 10 id, title FROM articles', $query->sql());
 	}
 
+/**
+ * Test select with limit on lte SQLServer2008
+ *
+ * @return void
+ */
+	public function testSelectLimitOldServer() {
+		$driver = $this->getMock(
+			'Cake\Database\Driver\Sqlserver',
+			['_connect', 'connection', '_version'],
+			[['dsn' => 'foo']]
+		);
+		$driver
+			->expects($this->any())
+			->method('_version')
+			->will($this->returnValue(8));
+
+		$connection = $this->getMock(
+			'\Cake\Database\Connection',
+			['connect', 'driver'],
+			[['log' => false]]
+		);
+		$connection
+			->expects($this->any())
+			->method('driver')
+			->will($this->returnValue($driver));
+
+		$query = new \Cake\Database\Query($connection);
+		$query->select(['id', 'title'])
+			->from('articles')
+			->limit(10);
+		$expected = 'SELECT TOP 10 id, title FROM articles';
+		$this->assertEquals($expected, $query->sql());
+
+		$query = new \Cake\Database\Query($connection);
+		$query->select(['id', 'title'])
+			->from('articles')
+			->offset(10);
+		$expected = 'SELECT * FROM (SELECT id, title, (ROW_NUMBER() OVER (ORDER BY (SELECT NULL))) AS [_cake_page_rownum_] ' .
+			'FROM articles) AS _cake_paging_ ' .
+			'WHERE _cake_paging_._cake_page_rownum_ > :c0';
+		$this->assertEquals($expected, $query->sql());
+
+		$query = new \Cake\Database\Query($connection);
+		$query->select(['id', 'title'])
+			->from('articles')
+			->order(['id'])
+			->offset(10);
+		$expected = 'SELECT * FROM (SELECT id, title, (ROW_NUMBER() OVER (ORDER BY id)) AS [_cake_page_rownum_] ' .
+			'FROM articles) AS _cake_paging_ ' .
+			'WHERE _cake_paging_._cake_page_rownum_ > :c0';
+		$this->assertEquals($expected, $query->sql());
+
+		$query = new \Cake\Database\Query($connection);
+		$query->select(['id', 'title'])
+			->from('articles')
+			->order(['id'])
+			->limit(10)
+			->offset(50);
+		$expected = 'SELECT * FROM (SELECT id, title, (ROW_NUMBER() OVER (ORDER BY id)) AS [_cake_page_rownum_] ' .
+			'FROM articles) AS _cake_paging_ ' .
+			'WHERE (_cake_paging_._cake_page_rownum_ > :c0 AND _cake_paging_._cake_page_rownum_ <= :c1)';
+		$this->assertEquals($expected, $query->sql());
+	}
+
 }