Merge pull request #200 from CartoDB/CDB-3686

Affected tables and last updated time for a query into a single SQL API request

NEWS.md

@@ -1,6 +1,13 @@
 1.13.2 -- 2014-mm-dd
 --------------------
 
+Enhancements:
+ - SQL API requests moved to its own entity
+
+New features:
+ - Affected tables and last updated time for a query are performed in a single request to the SQL API
+
+
 1.13.1 -- 2014-08-04
 --------------------
 

lib/cartodb/api/query_tables_api.js

@@ -0,0 +1,95 @@
+var sqlApi = require('../sql/sql_api');
+
+function QueryTablesApi() {
+}
+
+var affectedTableRegexCache = {
+    bbox: /!bbox!/g,
+    pixel_width: /!pixel_width!/g,
+    pixel_height: /!pixel_height!/g
+};
+
+module.exports = QueryTablesApi;
+
+QueryTablesApi.prototype.getLastUpdatedTime = function (username, api_key, tableNames, callback) {
+    var sql = 'SELECT EXTRACT(EPOCH FROM max(updated_at)) as max FROM CDB_TableMetadata m WHERE m.tabname = any (ARRAY['+
+        tableNames.map(function(t) { return "'" + t + "'::regclass"; }).join(',') +
+        '])';
+
+    // call sql api
+    sqlApi.query(username, api_key, sql, function(err, rows){
+        if (err){
+            var msg = err.message ? err.message : err;
+            callback(new Error('could not find last updated timestamp: ' + msg));
+            return;
+        }
+        // when the table has not updated_at means it hasn't been changed so a default last_updated is set
+        var last_updated = 0;
+        if(rows.length !== 0) {
+            last_updated = rows[0].max || 0;
+        }
+
+        callback(null, last_updated*1000);
+    });
+};
+
+QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, api_key, sql, callback) {
+    // Replace mapnik tokens
+    sql = sql
+        .replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
+        .replace(affectedTableRegexCache.pixel_width, '1')
+        .replace(affectedTableRegexCache.pixel_height, '1')
+    ;
+
+    // Pass to CDB_QueryTables
+    sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
+
+    // call sql api
+    sqlApi.query(username, api_key, sql, function(err, rows){
+        if (err){
+            var msg = err.message ? err.message : err;
+            callback(new Error('could not fetch source tables: ' + msg));
+            return;
+        }
+        var qtables = rows[0].cdb_querytables;
+        var tableNames = qtables.split(/^\{(.*)\}$/)[1];
+        tableNames = tableNames ? tableNames.split(',') : [];
+        callback(null, tableNames);
+    });
+};
+
+QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (username, api_key, sql, callback) {
+    sql = sql
+        .replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
+        .replace(affectedTableRegexCache.pixel_width, '1')
+        .replace(affectedTableRegexCache.pixel_height, '1')
+    ;
+
+    var query = [
+        'SELECT',
+            'CDB_QueryTables($windshaft$' + sql + '$windshaft$) as tablenames,',
+        'EXTRACT(EPOCH FROM max(updated_at)) as max',
+        'FROM CDB_TableMetadata m',
+            'WHERE m.tabname = any (CDB_QueryTables($windshaft$' + sql + '$windshaft$)::regclass[])'
+    ].join(' ');
+
+    sqlApi.query(username, api_key, query, function(err, rows){
+        if (err || rows.length === 0) {
+            var msg = err.message ? err.message : err;
+            callback(new Error('could not fetch affected tables and last updated time: ' + msg));
+            return;
+        }
+
+        var result = rows[0];
+
+        var tableNames = result.tablenames.split(/^\{(.*)\}$/)[1];
+        tableNames = tableNames ? tableNames.split(',') : [];
+
+        var lastUpdatedTime = result.max || 0;
+
+        callback(null, {
+            affectedTables: tableNames,
+            lastUpdatedTime: lastUpdatedTime * 1000
+        });
+    });
+};

lib/cartodb/cartodb_windshaft.js

@@ -2,7 +2,7 @@
 var _ = require('underscore')
     , Step       = require('step')
     , Windshaft = require('windshaft')
-    , redisPool = new require('redis-mpool')(global.environment.redis)
+    , redisPool = require('redis-mpool')(global.environment.redis)
     // TODO: instanciate cartoData with redisPool
     , cartoData  = require('cartodb-redis')(global.environment.redis)
     , SignedMaps = require('./signed_maps.js')

lib/cartodb/server_options.js

@@ -2,9 +2,9 @@ var   _          = require('underscore')
     , Step       = require('step')
     , cartoData  = require('cartodb-redis')(global.environment.redis)
     , Cache      = require('./cache_validator')
+    , QueryTablesApi   = require('./api/query_tables_api')
     , mapnik     = require('mapnik')
     , crypto     = require('crypto')
-    , request    = require('request')
     , LZMA       = require('lzma/lzma_worker.js').LZMA
   ;
 
@@ -19,6 +19,8 @@ if ( _.isUndefined(global.environment.sqlapi.domain) ) {
 
 module.exports = function(){
 
+    var queryTablesApi = new QueryTablesApi();
+
     var rendererConfig = _.defaults(global.environment.renderer || {}, {
       cache_ttl: 60000, // milliseconds
       metatile: 4,
@@ -88,121 +90,6 @@ module.exports = function(){
     //       we have no SQL after layer creation.
     me.channelCache = {};
 
-    // Run a query through the SQL api
-    me.sqlQuery = function (username, api_key, sql, callback) {
-      var api = global.environment.sqlapi;
-
-      // build up api string
-      var sqlapihostname = username;
-      if ( api.domain ) sqlapihostname += '.' + api.domain;
-
-      var sqlapi = api.protocol + '://';
-      if ( api.host && api.host != api.domain ) sqlapi += api.host;
-      else sqlapi += sqlapihostname;
-      sqlapi += ':' + api.port + '/api/' + api.version + '/sql';
-
-      var qs  = { q: sql }
-
-      // add api_key if given
-      if (_.isString(api_key) && api_key != '') { qs.api_key = api_key; }
-
-      // call sql api
-      //
-      // NOTE: using POST to avoid size limits:
-      // See http://github.com/CartoDB/Windshaft-cartodb/issues/111
-      //
-      // NOTE: uses "host" header to allow IP based specification
-      //       of sqlapi address (and avoid a DNS lookup)
-      //
-      // NOTE: allows for keeping up to "maxConnections" concurrent
-      //       sockets opened per SQL-API host.
-      // See http://nodejs.org/api/http.html#http_agent_maxsockets
-      //
-      var maxSockets = global.environment.maxConnections || 128;
-      var maxGetLen = api.max_get_sql_length || 2048;
-      var maxSQLTime = api.timeout || 100; // 1/10 of a second by default
-      var reqSpec = {
-            url:sqlapi,
-            json:true,
-            headers:{host: sqlapihostname}
-            // http://nodejs.org/api/http.html#http_agent_maxsockets
-            ,pool:{maxSockets:maxSockets}
-            // timeout in milliseconds
-            ,timeout:maxSQLTime
-      }
-      if ( sql.length > maxGetLen ) {
-        reqSpec.method = 'POST';
-        reqSpec.body = qs;
-      } else {
-        reqSpec.method = 'GET';
-        reqSpec.qs = qs;
-      }
-      request(reqSpec, function(err, res, body) {
-          if (err){
-            console.log('ERROR connecting to SQL API on ' + sqlapi + ': ' + err);
-            callback(err);
-            return;
-          }
-          if (res.statusCode != 200) {
-              var msg = res.body.error ? res.body.error : res.body;
-              callback(new Error(msg));
-              console.log('unexpected response status (' + res.statusCode + ') for sql query: ' + sql + ': ' + msg);
-              return;
-          }
-          callback(null, body.rows);
-      });
-    };
-
-    //
-    // Invoke callback with number of milliseconds since
-    // last update in any of the given tables
-    //
-    me.findLastUpdated = function (username, api_key, tableNames, callback) {
-      var sql = 'SELECT EXTRACT(EPOCH FROM max(updated_at)) as max FROM CDB_TableMetadata m WHERE m.tabname = any (ARRAY['+
-         tableNames.map(function(t) { return "'" + t + "'::regclass"; }).join(',') +
-      '])';
-
-      // call sql api
-      me.sqlQuery(username, api_key, sql, function(err, rows){
-          if (err){
-            var msg = err.message ? err.message : err;
-            callback(new Error('could not find last updated timestamp: ' + msg));
-            return;
-          }
-          // when the table has not updated_at means it hasn't been changed so a default last_updated is set
-          var last_updated = 0;
-          if(rows.length !== 0) {
-            last_updated = rows[0].max || 0;
-          }
-          callback(null, last_updated*1000);
-      });
-    };
-
-    me.affectedTables = function (username, api_key, sql, callback) {
-
-      // Replace mapnik tokens
-      sql = sql.replace(RegExp('!bbox!', 'g'), 'ST_MakeEnvelope(0,0,0,0)')
-               .replace(RegExp('!pixel_width!', 'g'), '1')
-               .replace(RegExp('!pixel_height!', 'g'), '1')
-            ;
-
-      // Pass to CDB_QueryTables
-      sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
-
-      // call sql api
-      me.sqlQuery(username, api_key, sql, function(err, rows){
-          if (err){
-            var msg = err.message ? err.message : err;
-            callback(new Error('could not fetch source tables: ' + msg));
-            return;
-          }
-          var qtables = rows[0].cdb_querytables;
-          var tableNames = qtables.split(/^\{(.*)\}$/)[1];
-          tableNames = tableNames ? tableNames.split(',') : [];
-          callback(null, tableNames);
-      });
-    };
-
     me.buildCacheChannel = function (dbName, tableNames){
       return dbName + ':' + tableNames.join(',');
     };
@@ -211,7 +98,7 @@ module.exports = function(){
         var hash = crypto.createHash('md5');
         hash.update(data);
         return hash.digest('hex');
-    }
+    };
 
     me.generateCacheChannel = function(app, req, callback){
 
@@ -241,7 +128,6 @@ module.exports = function(){
             // See http://github.com/CartoDB/Windshaft-cartodb/issues/152
             if ( ! app.mapStore ) {
               throw new Error('missing channel cache for token ' + req.params.token);
-              return;
             }
             var next = this;
             var mapStore = app.mapStore;
@@ -305,7 +191,7 @@ module.exports = function(){
                 if ( req.profiler ) req.profiler.done('getSignerMapKey');
                 key = data;
               }
-              me.affectedTables(user, key, sql, this); // in addCacheChannel
+              queryTablesApi.getAffectedTablesInQuery(user, key, sql, this); // in addCacheChannel
             },
             function finish(err, data) {
               next(err,data);
@@ -397,7 +283,7 @@ module.exports = function(){
             err = errors.length ? new Error(errors.join('\n')) : null;
             callback(err);
           }
-        }
+        };
 
         // include in layergroup response the variables in serverMedata
         // those variables are useful to send to the client information
@@ -427,18 +313,17 @@ module.exports = function(){
         var key    = req.params.map_key || req.params.api_key;
 
         var cacheKey = dbName + ':' + token;
-        var tabNames;
 
         Step(
-          function getTables() {
+            function getAffectedTablesAndLastUpdatedTime() {
-            me.affectedTables(usr, key, sql, this); // in afterLayergroupCreate
+                queryTablesApi.getAffectedTablesAndLastUpdatedTime(usr, key, sql, this);
             },
-          function getLastupdated(err, tableNames) {
+            function handleAffectedTablesAndLastUpdatedTime(err, result) {
-            if (req.profiler) req.profiler.done('affectedTables');
+                if (req.profiler) req.profiler.done('queryTablesAndLastUpdated');
                 if ( err ) throw err;
-            var cacheChannel = me.buildCacheChannel(dbName,tableNames);
+                var cacheChannel = me.buildCacheChannel(dbName, result.affectedTables);
-            // store for caching from me.afterLayergroupCreate
                 me.channelCache[cacheKey] = cacheChannel;
+
                 if (req.res && req.method == 'GET') {
                     var res = req.res;
                     if ( req.query && req.query.cache_policy == 'persist' ) {
@@ -450,16 +335,10 @@ module.exports = function(){
                     res.header('Last-Modified', (new Date()).toUTCString());
                     res.header('X-Cache-Channel', cacheChannel);
                 }
-            // find last updated
+
-            if ( ! tableNames.length ) return 0; // skip for no affected tables
+                // last update for layergroup cache buster
-            tabNames = tableNames;
+                response.layergroupid = response.layergroupid + ':' + result.lastUpdatedTime;
-            me.findLastUpdated(usr, key, tableNames, this);
+                response.last_updated = new Date(result.lastUpdatedTime).toISOString();
-          },
-          function(err, lastUpdated) {
-            if ( err ) throw err;
-            if (req.profiler && tabNames) req.profiler.done('findLastUpdated');
-            response.layergroupid = response.layergroupid + ':' + lastUpdated; // use epoch
-            response.last_updated = new Date(lastUpdated).toISOString(); 
                 return null;
             },
             function finish(err) {
@@ -490,7 +369,7 @@ module.exports = function(){
         return;
       }
       return mat[1];
-    }
+    };
 
     // Set db authentication parameters to those of the given username
     //
@@ -742,7 +621,7 @@ module.exports = function(){
           //console.log("type of req.query.lzma is " + typeof(req.query.lzma));
 
           // Decode (from base64)
-          var lzma = (new Buffer(req.query.lzma, 'base64').toString('binary')).split('').map(function(c) { return c.charCodeAt(0) - 128 })
+          var lzma = (new Buffer(req.query.lzma, 'base64').toString('binary')).split('').map(function(c) { return c.charCodeAt(0) - 128 });
 
           // Decompress
           LZMA.decompress(
@@ -750,8 +629,8 @@ module.exports = function(){
             function(result) {
               if (req.profiler) req.profiler.done('LZMA decompress');
               try {
-                delete req.query.lzma
+                delete req.query.lzma;
-                _.extend(req.query, JSON.parse(result))
+                _.extend(req.query, JSON.parse(result));
                 me.req2params(req, callback);
               } catch (err) {
                 callback(new Error('Error parsing lzma as JSON: ' + err));
@@ -783,7 +662,7 @@ module.exports = function(){
             req.params.signer = tksplit.shift();
             if ( ! req.params.signer ) req.params.signer = user;
             else if ( req.params.signer != user ) {
-              var err = new Error('Cannot use map signature of user "' + req.params.signer + '" on database of user "' + user + '"')
+              var err = new Error('Cannot use map signature of user "' + req.params.signer + '" on database of user "' + user + '"');
               err.http_status = 403;
               callback(err);
               return;

lib/cartodb/sql/sql_api.js

@@ -0,0 +1,66 @@
+var _       = require('underscore'),
+    request = require('request');
+
+module.exports.query = function (username, api_key, sql, callback) {
+    var api = global.environment.sqlapi;
+
+    // build up api string
+    var sqlapihostname = username;
+    if ( api.domain ) sqlapihostname += '.' + api.domain;
+
+    var sqlapi = api.protocol + '://';
+    if ( api.host && api.host != api.domain ) sqlapi += api.host;
+    else sqlapi += sqlapihostname;
+    sqlapi += ':' + api.port + '/api/' + api.version + '/sql';
+
+    var qs  = { q: sql };
+
+    // add api_key if given
+    if (_.isString(api_key) && api_key != '') { qs.api_key = api_key; }
+
+    // call sql api
+    //
+    // NOTE: using POST to avoid size limits:
+    // See http://github.com/CartoDB/Windshaft-cartodb/issues/111
+    //
+    // NOTE: uses "host" header to allow IP based specification
+    //       of sqlapi address (and avoid a DNS lookup)
+    //
+    // NOTE: allows for keeping up to "maxConnections" concurrent
+    //       sockets opened per SQL-API host.
+    // See http://nodejs.org/api/http.html#http_agent_maxsockets
+    //
+    var maxSockets = global.environment.maxConnections || 128;
+    var maxGetLen = api.max_get_sql_length || 2048;
+    var maxSQLTime = api.timeout || 100; // 1/10 of a second by default
+    var reqSpec = {
+        url:sqlapi,
+        json:true,
+        headers:{host: sqlapihostname}
+        // http://nodejs.org/api/http.html#http_agent_maxsockets
+        ,pool:{maxSockets:maxSockets}
+        // timeout in milliseconds
+        ,timeout:maxSQLTime
+    };
+    if ( sql.length > maxGetLen ) {
+        reqSpec.method = 'POST';
+        reqSpec.body = qs;
+    } else {
+        reqSpec.method = 'GET';
+        reqSpec.qs = qs;
+    }
+    request(reqSpec, function(err, res, body) {
+        if (err){
+            console.log('ERROR connecting to SQL API on ' + sqlapi + ': ' + err);
+            callback(err);
+            return;
+        }
+        if (res.statusCode != 200) {
+            var msg = res.body.error ? res.body.error : res.body;
+            callback(new Error(msg));
+            console.log('unexpected response status (' + res.statusCode + ') for sql query: ' + sql + ': ' + msg);
+            return;
+        }
+        callback(null, body.rows);
+    });
+};

test/acceptance/multilayer.js

@@ -111,10 +111,14 @@ suite('multilayer', function() {
               assert.equal(cc.substring(0, dbname.length), dbname);
               var jsonquery = cc.substring(dbname.length+1);
               var sentquery = JSON.parse(jsonquery);
+              var expectedQuery = [layergroup.layers[0].options.sql, ';', layergroup.layers[1].options.sql].join('');
               assert.equal(sentquery.q, 'SELECT CDB_QueryTables($windshaft$'
-                + layergroup.layers[0].options.sql + ';'
+                + expectedQuery
-                + layergroup.layers[1].options.sql 
+                + '$windshaft$) as tablenames, EXTRACT(EPOCH FROM max(updated_at)) as max'
-                + '$windshaft$)');
+                + ' FROM CDB_TableMetadata m'
+                + ' WHERE m.tabname = any (CDB_QueryTables($windshaft$'
+                + expectedQuery
+                + '$windshaft$)::regclass[])');
 
               assert.imageEqualsFile(res.body, 'test/fixtures/test_table_0_0_0_multilayer1.png', IMAGE_EQUALS_HIGHER_TOLERANCE_PER_MIL,
                 function(err, similarity) {
@@ -387,12 +391,17 @@ suite('multilayer', function() {
               assert.equal(cc.substring(0, dbname.length), dbname);
               var jsonquery = cc.substring(dbname.length+1);
               var sentquery = JSON.parse(jsonquery);
+              var expectedQuery = layergroup.layers[0].options.sql
+                  .replace(/!bbox!/g, 'ST_MakeEnvelope(0,0,0,0)')
+                  .replace(/!pixel_width!/g, '1')
+                  .replace(/!pixel_height!/g, '1');
               assert.equal(sentquery.q, 'SELECT CDB_QueryTables($windshaft$'
-                + layergroup.layers[0].options.sql
+                  + expectedQuery
-                    .replace(RegExp('!bbox!', 'g'), 'ST_MakeEnvelope(0,0,0,0)')
+                  + '$windshaft$) as tablenames, EXTRACT(EPOCH FROM max(updated_at)) as max'
-                    .replace(RegExp('!pixel_width!', 'g'), '1')
+                  + ' FROM CDB_TableMetadata m'
-                    .replace(RegExp('!pixel_height!', 'g'), '1')
+                  + ' WHERE m.tabname = any (CDB_QueryTables($windshaft$'
-                + '$windshaft$)');
+                  + expectedQuery
+                  + '$windshaft$)::regclass[])');
 
               assert.imageEqualsFile(res.body, 'test/fixtures/test_multilayer_bbox.png', IMAGE_EQUALS_TOLERANCE_PER_MIL,
                 function(err, similarity) {
@@ -420,12 +429,17 @@ suite('multilayer', function() {
               assert.equal(cc.substring(0, dbname.length), dbname);
               var jsonquery = cc.substring(dbname.length+1);
               var sentquery = JSON.parse(jsonquery);
-              assert.equal(sentquery.q, 'SELECT CDB_QueryTables($windshaft$'
+              var expectedQuery = layergroup.layers[0].options.sql
-                + layergroup.layers[0].options.sql
                   .replace('!bbox!', 'ST_MakeEnvelope(0,0,0,0)')
                   .replace('!pixel_width!', '1')
-                    .replace('!pixel_height!', '1')
+                  .replace('!pixel_height!', '1');
-                + '$windshaft$)');
+              assert.equal(sentquery.q, 'SELECT CDB_QueryTables($windshaft$'
+                + expectedQuery
+                + '$windshaft$) as tablenames, EXTRACT(EPOCH FROM max(updated_at)) as max'
+                  + ' FROM CDB_TableMetadata m'
+                  + ' WHERE m.tabname = any (CDB_QueryTables($windshaft$'
+                  + expectedQuery
+                  + '$windshaft$)::regclass[])');
 
               assert.imageEqualsFile(res.body, 'test/fixtures/test_multilayer_bbox.png', IMAGE_EQUALS_TOLERANCE_PER_MIL,
                 function(err, similarity) {

test/support/SQLAPIEmu.js

@@ -2,7 +2,7 @@ var http        = require('http');
 var url         = require('url');
 var _ = require('underscore');
 
-var o = function(port, cb) {
+var SQLAPIEmulator = function(port, cb) {
 
   this.queries = [];
   var that = this;
@@ -37,7 +37,7 @@ var o = function(port, cb) {
    }).listen(port, cb);
 };
 
-o.prototype.handleQuery = function(query, res) {
+SQLAPIEmulator.prototype.handleQuery = function(query, res) {
     this.queries.push(query);
     if ( query.q.match('SQLAPIERROR') ) {
       res.statusCode = 400;
@@ -45,39 +45,37 @@ o.prototype.handleQuery = function(query, res) {
     } else if ( query.q.match('SQLAPINOANSWER') ) {
         console.log("SQLAPIEmulator will never respond, on request");
         return;
+    } else if (query.q.match('tablenames')) {
+        var tableNames = JSON.stringify(query);
+        res.write(queryResult({tablenames: '{' + tableNames + '}', max: 1234567890.123}));
     } else if ( query.q.match('EPOCH.* as max') ) {
       // This is the structure of the known query sent by tiler
-      var row = {
+        res.write(queryResult({max: 1234567890.123}));
-        'max': 1234567890.123
-      };
-      res.write(JSON.stringify({rows: [ row ]}));
     } else {
       if ( query.q.match('_private_') && query.api_key === undefined) {
         res.statusCode = 403;
         res.write(JSON.stringify({'error':'forbidden: ' + JSON.stringify(query)}));
       } else {
         var qs = JSON.stringify(query);
-        var row = {
+        res.write(queryResult({cdb_querytables: '{' + qs + '}', max: 1234567890.123}));
-          // This is the structure of the known query sent by tiler
-          'cdb_querytables': '{' + qs + '}',
-          'max': qs
-        };
-        var out_obj = {rows: [ row ]};
-        var out = JSON.stringify(out_obj);
-        res.write(out);
       }
     }
     res.end();
   };
 
-
+SQLAPIEmulator.prototype.close = function(cb) {
-o.prototype.close = function(cb) {
   this.sqlapi_server.close(cb);
 };
 
-o.prototype.getLastRequest = function() {
+SQLAPIEmulator.prototype.getLastRequest = function() {
   return this.requests.pop();
 };
 
-module.exports = o;
+function queryResult(row) {
+    return JSON.stringify({
+        rows: [row]
+    });
+}
+
+module.exports = SQLAPIEmulator;