Require exact counts for determining the starting datasource in CompositeDatasource

Author: rubensworks

lib/datasources/CompositeDatasource.js

@@ -51,6 +51,20 @@ CompositeDatasource.prototype._getDatasourceById = function(datasourceIndex) {
   return this._datasources[this._datasourceNames[datasourceIndex]].datasource;
 };
 
+// Count the amount of triple in the query result to get an exact count.
+CompositeDatasource.prototype._getExactCount = function(datasource, query, cb) {
+  var emptyQuery = { offset: 0, subject: query.subject, predicate: query.predicate, object: query.object };
+  var exactCount = 0;
+  var countingTripleStream = { push: function(triple) {
+    if (triple) {
+      exactCount++;
+    } else {
+      cb(exactCount);
+    }
+  }};
+  datasource._executeQuery(emptyQuery, countingTripleStream, noop);
+};
+
 // Recursively find all required datasource composition info to perform a query.
 // The callback will provide the parameters:
 //   Datasource id to start querying from
@@ -70,21 +84,35 @@ CompositeDatasource.prototype._getDatasourceInfo = function(query, absoluteOffse
       cb(chosenDatasource, chosenOffset, totalCount, exactCount);
     } else {
       var emptyTripleStream = { push: noop };
-      self._getDatasourceById(datasourceIndex)._executeQuery(emptyQuery, emptyTripleStream, function (metadata) {
+      var datasource = self._getDatasourceById(datasourceIndex);
+      datasource._executeQuery(emptyQuery, emptyTripleStream, function (metadata) {
         var count = metadata.totalCount;
         var exact = metadata.exactCount;
-        if (chosenDatasource < 0 && offset < count) {
-          // We can start querying from this datasource
-          setImmediate(function () {
-            findRecursive(datasourceIndex + 1, offset - count, datasourceIndex, offset,
-                totalCount + count, exactCount && exact);
+        // If we are still looking for an appropriate datasource, we need exact counts!
+        if (offset > 0 && !exact) {
+          self._getExactCount(datasource, query, function(exactCount) {
+            count = exactCount;
+            exact = true;
+            continueRecursion();
           });
         } else {
-          // We forward our accumulated information and go check the next datasource
-          setImmediate(function () {
-            findRecursive(datasourceIndex + 1, offset - count, chosenDatasource, chosenOffset,
-                totalCount + count, exactCount && exact);
-          });
+          continueRecursion();
+        }
+
+        function continueRecursion() {
+          if (chosenDatasource < 0 && offset < count) {
+            // We can start querying from this datasource
+            setImmediate(function () {
+              findRecursive(datasourceIndex + 1, offset - count, datasourceIndex, offset,
+                  totalCount + count, exactCount && exact);
+            });
+          } else {
+            // We forward our accumulated information and go check the next datasource
+            setImmediate(function () {
+              findRecursive(datasourceIndex + 1, offset - count, chosenDatasource, chosenOffset,
+                  totalCount + count, exactCount && exact);
+            });
+          }
         }
       });
     }