Emit whether or not the triple counts are exact in the metadata callback

Author: rubensworks

lib/datasources/CompositeDatasource.js

@@ -62,25 +62,28 @@ CompositeDatasource.prototype._getDatasourceInfo = function(query, absoluteOffse
     offset: 0, limit: 1,
     subject: query.subject, predicate: query.predicate, object: query.object
   };
-  return findRecursive(0, absoluteOffset, -1, -1, 0, cb);
+  return findRecursive(0, absoluteOffset, -1, -1, 0, cb, true);
 
-  function findRecursive(datasourceIndex, offset, chosenDatasource, chosenOffset, totalCount) {
+  function findRecursive(datasourceIndex, offset, chosenDatasource, chosenOffset, totalCount, exactCount) {
     if (datasourceIndex >= self._datasourceNames.length) {
       // We checked all datasources, return our accumulated information
-      cb(chosenDatasource, chosenOffset, totalCount);
+      cb(chosenDatasource, chosenOffset, totalCount, exactCount);
     } else {
       var emptyTripleStream = { push: noop };
       self._getDatasourceById(datasourceIndex)._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);
+            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);
+            findRecursive(datasourceIndex + 1, offset - count, chosenDatasource, chosenOffset,
+                totalCount + count, exactCount && exact);
           });
         }
       });
@@ -94,14 +97,14 @@ function noop() {}
 CompositeDatasource.prototype._executeQuery = function (query, tripleStream, metadataCallback) {
   var offset =  query.offset || 0, limit = query.limit || Infinity;
   var self = this;
-  this._getDatasourceInfo(query, offset, function(datasourceIndex, relativeOffset, totalCount) {
+  this._getDatasourceInfo(query, offset, function(datasourceIndex, relativeOffset, totalCount, exactCount) {
     if (datasourceIndex < 0) {
       // No valid datasource has been found
-      metadataCallback({ totalCount: totalCount });
+      metadataCallback({ totalCount: totalCount, exactCount: exactCount });
       tripleStream.push(null);
     } else {
       // Send query to first applicable datasource and optionally emit triples from consecutive datasources
-      metadataCallback({ totalCount: totalCount });
+      metadataCallback({ totalCount: totalCount, exactCount: exactCount });
       var emitted = 0;
 
       // Modify our triple stream so that if all results from one datasource have arrived,

lib/datasources/ExternalHdtDatasource.js

@@ -53,7 +53,7 @@ ExternalHdtDatasource.prototype._executeQuery = function (query, tripleStream, m
       // Ensure the estimated total count is as least as large as the number of triples
       if (tripleCount && estimatedTotalCount < offset + tripleCount)
         estimatedTotalCount = offset + (tripleCount < query.limit ? tripleCount : 2 * tripleCount);
-      metadataCallback({ totalCount: estimatedTotalCount });
+      metadataCallback({ totalCount: estimatedTotalCount, exactCount: false });
       tripleStream.push(null);
     }
   });

lib/datasources/HdtDatasource.js

@@ -33,13 +33,13 @@ HdtDatasource.prototype._initialize = function (done) {
 HdtDatasource.prototype._executeQuery = function (query, tripleStream, metadataCallback) {
   this._hdtDocument.search(query.subject, query.predicate, query.object,
                            { limit: query.limit, offset: query.offset },
-    function (error, triples, estimatedTotalCount) {
+    function (error, triples, estimatedTotalCount, exactCount) {
       if (error) return tripleStream.emit('error', error);
       // Ensure the estimated total count is as least as large as the number of triples
       var tripleCount = triples.length, offset = query.offset || 0;
       if (tripleCount && estimatedTotalCount < offset + tripleCount)
         estimatedTotalCount = offset + (tripleCount < query.limit ? tripleCount : 2 * tripleCount);
-      metadataCallback({ totalCount: estimatedTotalCount });
+      metadataCallback({ totalCount: estimatedTotalCount, exactCount: exactCount });
       // Add the triples to the stream
       for (var i = 0; i < tripleCount; i++)
         tripleStream.push(triples[i]);

lib/datasources/MemoryDatasource.js

@@ -29,7 +29,7 @@ MemoryDatasource.prototype._executeQuery = function (query, tripleStream, metada
   var offset =  query.offset || 0, limit = query.limit || Infinity,
       triples = this._tripleStore.findByIRI(query.subject, query.predicate, query.object);
   // Send the metadata
-  metadataCallback({ totalCount: triples.length });
+  metadataCallback({ totalCount: triples.length, exactCount: true });
   // Send the requested subset of triples
   for (var i = offset, l = Math.min(offset + limit, triples.length); i < l; i++)
     tripleStream.push(triples[i]);

lib/datasources/SparqlDatasource.js

@@ -55,7 +55,7 @@ SparqlDatasource.prototype._executeQuery = function (query, tripleStream, metada
   // Determine the total number of matching triples
   this._getPatternCount(sparqlPattern, function (error, totalCount) {
     if (error) emitError(error);
-    else if (typeof totalCount === 'number') metadataCallback({ totalCount: totalCount });
+    else if (typeof totalCount === 'number') metadataCallback({ totalCount: totalCount, exactCount: true });
   });
 
   // Emits an error on the triple stream