Implement an Overviews query rewriter

Use the Windshaft query-rewriter interface to adapt queries so
they use available overview tables.

This requires a version of Windshaft that implements the query-rewriter
interface (package.json/npm-shrinkwap.json have yet to be updated)

lib/cartodb/api/overviews_metadata_api.js

@@ -1,11 +1,10 @@
-// TODO: use pgQueryRunner parameter directly
-function OverviewsApi(pgQueryRunner) {
+function OverviewsMetadataApi(pgQueryRunner) {
    this.pgQueryRunner = pgQueryRunner;
 }
 
-module.exports = OverviewsApi;
+module.exports = OverviewsMetadataApi;
 
-// TODO: share this with OverviewsApi? ... or maintain independence?
+// TODO: share this with QueryTablesApi? ... or maintain independence?
 var affectedTableRegexCache = {
     bbox: /!bbox!/g,
     scale_denominator: /!scale_denominator!/g,
@@ -22,7 +21,7 @@ function prepareSql(sql) {
     ;
 }
 
-OverviewsApi.prototype.getOverviewsMetadata = function (username, sql, callback) {
+OverviewsMetadataApi.prototype.getOverviewsMetadata = function (username, sql, callback) {
     var query = 'SELECT * FROM CDB_Overviews(CDB_QueryTablesText($windshaft$' + prepareSql(sql) + '$windshaft$))';
     this.pgQueryRunner.run(username, query, function handleOverviewsRows(err, rows) {
         if (err){

lib/cartodb/controllers/map.js

@@ -33,7 +33,7 @@ var MapConfigOverviewsAdapter = require('../models/mapconfig_overviews_adapter')
  * @constructor
  */
 function MapController(authApi, pgConnection, templateMaps, mapBackend, metadataBackend,
-                       queryTablesApi, overviewsApi,
+                       queryTablesApi, overviewsMetadataApi,
                        surrogateKeysCache, userLimitsApi, layergroupAffectedTables) {
 
     BaseController.call(this, authApi, pgConnection);
@@ -43,13 +43,13 @@ function MapController(authApi, pgConnection, templateMaps, mapBackend, metadata
     this.mapBackend = mapBackend;
     this.metadataBackend = metadataBackend;
     this.queryTablesApi = queryTablesApi;
-    this.overviewsApi = overviewsApi;
+    this.overviewsMetadataApi = overviewsMetadataApi;
     this.surrogateKeysCache = surrogateKeysCache;
     this.userLimitsApi = userLimitsApi;
     this.layergroupAffectedTables = layergroupAffectedTables;
 
     this.namedLayersAdapter = new MapConfigNamedLayersAdapter(templateMaps);
-    this.overviewsAdapter = new MapConfigOverviewsAdapter(this.overviewsApi);
+    this.overviewsAdapter = new MapConfigOverviewsAdapter(this.overviewsMetadataApi);
 }
 
 util.inherits(MapController, BaseController);

lib/cartodb/models/mapconfig_overviews_adapter.js

@@ -1,8 +1,8 @@
 var queue = require('queue-async');
 var _ = require('underscore');
 
-function MapConfigOverviewsAdapter(overviewsApi) {
-    this.overviewsApi = overviewsApi;
+function MapConfigOverviewsAdapter(overviewsMetadataApi) {
+    this.overviewsMetadataApi = overviewsMetadataApi;
 }
 
 module.exports = MapConfigOverviewsAdapter;
@@ -20,13 +20,13 @@ MapConfigOverviewsAdapter.prototype.getLayers = function(username, layers, callb
       if ( layer.type !== 'mapnik' && layer.type !== 'cartodb' ) {
           return done(null, layer);
       }
-      self.overviewsApi.getOverviewsMetadata(username, layer.options.sql, function(err, metadata){
+      self.overviewsMetadataApi.getOverviewsMetadata(username, layer.options.sql, function(err, metadata){
            if (err) {
                done(err, layer);
            } else {
                if ( !_.isEmpty(metadata) ) {
                    layer = _.extend({}, layer);
-                   layer.options = _.extend({}, layer.options, { overviews: metadata });
+                   layer.options = _.extend({}, layer.options, { query_rewrite_data: { overviews: metadata } });
                }
                done(null, layer);
            }

lib/cartodb/server.js

@@ -20,7 +20,7 @@ var mapnik = windshaft.mapnik;
 
 var TemplateMaps = require('./backends/template_maps.js');
 var QueryTablesApi = require('./api/query_tables_api');
-var OverviewsApi = require('./api/overviews_api');
+var OverviewsMetadataApi = require('./api/overviews_metadata_api');
 var UserLimitsApi = require('./api/user_limits_api');
 var AuthApi = require('./api/auth_api');
 var LayergroupAffectedTablesCache = require('./cache/layergroup_affected_tables');
@@ -53,7 +53,7 @@ module.exports = function(serverOptions) {
     var pgConnection = new PgConnection(metadataBackend);
     var pgQueryRunner = new PgQueryRunner(pgConnection);
     var queryTablesApi = new QueryTablesApi(pgQueryRunner);
-    var overviewsApi = new OverviewsApi(pgQueryRunner);
+    var overviewsMetadataApi = new OverviewsMetadataApi(pgQueryRunner);
     var userLimitsApi = new UserLimitsApi(metadataBackend, {
         limits: {
             cacheOnTimeout: serverOptions.renderer.mapnik.limits.cacheOnTimeout || false,
@@ -177,7 +177,7 @@ module.exports = function(serverOptions) {
         mapBackend,
         metadataBackend,
         queryTablesApi,
-        overviewsApi,
+        overviewsMetadataApi,
         surrogateKeysCache,
         userLimitsApi,
         layergroupAffectedTablesCache

lib/cartodb/server_options.js

@@ -1,9 +1,8 @@
 var os = require('os');
 var _ = require('underscore');
-var windshaft = require('windshaft');
-var OverviewsHandler = windshaft.OverviewsHandler;
+var OverviewsQueryRewriter = require('./utils/overviews_query_rewriter');
 
-var overviewsHandler = new OverviewsHandler({
+var overviewsQueryRewriter = new OverviewsQueryRewriter({
   zoom_level: 'CDB_ZoomFromScale(!scale_denominator!)'
 });
 
@@ -21,7 +20,7 @@ var rendererConfig = _.defaults(global.environment.renderer || {}, {
     http: {}
 });
 
-rendererConfig.mapnik.overviewsHandler = overviewsHandler;
+rendererConfig.mapnik.queryRewriter = overviewsQueryRewriter;
 
 // Perform keyword substitution in statsd
 // See https://github.com/CartoDB/Windshaft-cartodb/issues/153

lib/cartodb/utils/overviews_query_rewriter.js

@@ -0,0 +1,197 @@
+var TableNameParser = require('./table_name_parser');
+
+function OverviewsQueryRewriter(options) {
+
+    this.options = options;
+}
+
+module.exports = OverviewsQueryRewriter;
+
+// TODO: some names are introudced in the queries, and the
+// '_vovw_' (for vector overviews) is used in them, but no check
+// is performed for conflicts with existing identifiers in the query.
+
+// Build UNION expression to replace table, using overviews metadata
+// overviews metadata: { 1: 'table_ov1', ... }
+// assume table and overview names include schema if necessary and are quoted as needed
+function overviews_view_for_table(table, overviews_metadata, indent) {
+    var condition, i, len, ov_table, overview_layers, selects, z_hi, z_lo;
+    var parsed_table = TableNameParser.parse(table);
+
+    var sorted_overviews = []; // [[1, 'table_ov1'], ...]
+
+    indent = indent || '    ';
+    for (var z in overviews_metadata) {
+        if (overviews_metadata.hasOwnProperty(z)) {
+            sorted_overviews.push([z, overviews_metadata[z].table]);
+        }
+    }
+    sorted_overviews.sort(function(a, b){ return a[0]-b[0]; });
+
+    overview_layers = [];
+    z_lo = null;
+    for (i = 0, len = sorted_overviews.length; i < len; i++) {
+        z_hi = parseInt(sorted_overviews[i][0]);
+        ov_table = sorted_overviews[i][1];
+        overview_layers.push([overview_z_condition(z_lo, z_hi), ov_table]);
+        z_lo = z_hi;
+    }
+    overview_layers.push(["_vovw_z > " + z_lo, table]);
+
+    selects = overview_layers.map(function(condition_table) {
+      condition = condition_table[0];
+      ov_table = TableNameParser.parse(condition_table[1]);
+      ov_table.schema = ov_table.schema || parsed_table.schema;
+      var ov_identifier = TableNameParser.table_identifier(ov_table);
+      return indent + "SELECT * FROM " + ov_identifier + ", _vovw_scale WHERE " + condition;
+    });
+
+    return selects.join("\n"+indent+"UNION ALL\n");
+}
+
+function overview_z_condition(z_lo, z_hi) {
+    if (z_lo !== null) {
+        if (z_lo === z_hi - 1) {
+            return "_vovw_z = " + z_hi;
+        } else {
+            return "_vovw_z > " + z_lo + " AND _vovw_z <= " + z_hi;
+        }
+    } else {
+        if (z_hi === 0) {
+            return "_vovw_z = " + z_hi;
+        } else {
+            return "_vovw_z <= " + z_hi;
+        }
+    }
+}
+
+// name to be used for the view of the table using overviews
+function overviews_view_name(table) {
+    var parsed_table = TableNameParser.parse(table);
+    parsed_table.table = '_vovw_' + parsed_table.table;
+    parsed_table.schema = null;
+    return TableNameParser.table_identifier(parsed_table);
+}
+
+// replace a table name in a query by anoter name
+function replace_table_in_query(sql, old_table_name, new_table_name) {
+    var old_table = TableNameParser.parse(old_table_name);
+    var new_table = TableNameParser.parse(new_table_name);
+    var old_table_ident = TableNameParser.table_identifier(old_table);
+    var new_table_ident = TableNameParser.table_identifier(new_table);
+
+    // text that will be substituted by the table name pattern
+    var replacement = new_table_ident;
+
+    // regular expression prefix (beginning) to match a table name
+    function pattern_prefix(schema, identifier) {
+        if ( schema ) {
+            // to match a table name including schema prefix
+            // name should not be part of another name, so we require
+            // to start a at a word boundary
+            if ( identifier[0] !== '"' ) {
+                return '\\b';
+            } else {
+                return '';
+            }
+        } else {
+           // to match a table name without schema
+           // name should not begin right after a dot (i.e. have a explicit schema)
+           // nor be part of another name
+           // since the pattern matches the first character of the table
+           // it must be put back in the replacement text
+           replacement = '$01'+replacement;
+           return '([^\.a-z0-9_]|^)';
+        }
+    }
+
+    // regular expression suffix (ending) to match a table name
+    function pattern_suffix(identifier) {
+        // name shouldn't be the prefix of a longer name
+        if ( identifier[identifier.length-1] !== '"' ) {
+            return '\\b';
+        } else {
+            return '';
+        }
+    }
+
+    // regular expression to match a table name
+    var regexp = pattern_prefix(old_table.schema, old_table_ident) +
+                 old_table_ident +
+                 pattern_suffix(old_table_ident);
+
+    // replace all occurrences of the table pattern
+    return sql.replace(new RegExp(regexp, 'g'), replacement);
+}
+
+function overviews_query(query, overviews, zoom_level_expression) {
+    var replaced_query = query;
+    var sql = "WITH\n  _vovw_scale AS ( SELECT " + zoom_level_expression + " AS _vovw_z )";
+    for ( var table in overviews ) {
+        if (overviews.hasOwnProperty(table)) {
+            var table_overviews = overviews[table];
+            var table_view = overviews_view_name(table);
+            replaced_query = replace_table_in_query(replaced_query, table, table_view);
+            sql += ",\n  " + table_view + " AS (\n" + overviews_view_for_table(table, table_overviews) + "\n  )";
+        }
+    }
+    if ( replaced_query !== query ) {
+        sql += "\n";
+        sql += replaced_query;
+    } else {
+        sql = query;
+    }
+    return sql;
+}
+
+// Transform an SQL query so that it uses overviews.
+// overviews contains metadata about the overviews to be used:
+//     { 'table-name': {1: { table: 'overview-table-1' }, ... }, ... }
+//
+// For a given query `SELECT * FROM table`,  if any of tables in it
+// has overviews as defined by the provided metadat, the query will
+// be transform into something similar to this:
+//
+//     WITH _vovw_scale AS ( ... ), -- define scale level
+//     WITH _vovw_table AS ( ... ), -- define union of overviews and base table
+//     SELECT * FROM _vovw_table -- query with table replaced by _vovw_table
+//
+// This transformation can in principle be applied to arbitrary queries
+// (except for the case of queries that include the name of tables with
+// overviews inside text literals: at the current table name substitution
+// doesnn't prevent substitution inside literals).
+// But the transformation will currently only be applied to simple queries
+// of the form detected by the overviews_supported_query function.
+OverviewsQueryRewriter.prototype.query = function(query, data) {
+    var overviews = this.overviews_metadata(data);
+    if ( !overviews || !this.is_supported_query(query)) {
+        return query;
+    }
+    var zoom_level_expression = this.options.zoom_level || '0';
+    return overviews_query(query, overviews, zoom_level_expression);
+};
+
+OverviewsQueryRewriter.prototype.style = function(cartocss, cartocss_version, data) {
+    var overviews = this.overviews_metadata(data);
+    if ( !overviews ) {
+        return cartocss;
+    }
+    for ( var table in overviews ) {
+        if (overviews.hasOwnProperty(table)) {
+            // No modification of the CartoCSS is currently performed
+            // var table_view = overviews_view_name(table);
+            // cartocss = replace_table_in_style(cartocss, table, table_view);
+        }
+    }
+    return cartocss;
+};
+
+OverviewsQueryRewriter.prototype.is_supported_query = function(sql) {
+  return !!sql.match(
+    /^\s*SELECT\s+[\*\.a-z0-9_,\s]+?\s+FROM\s+((\"[^"]+\"|[a-z0-9_]+)\.)?(\"[^"]+\"|[a-z0-9_]+)\s*;?\s*$/i
+  );
+};
+
+OverviewsQueryRewriter.prototype.overviews_metadata = function(data) {
+  return data && data.overviews;
+};

lib/cartodb/utils/table_name_parser.js

@@ -0,0 +1,106 @@
+// Quote an PostgreSQL identifier if ncecessary
+function quote_identifier_if_needed(txt) {
+    if ( txt && !txt.match(/^[a-z_][a-z_0-9]*$/)) {
+        return '"' + txt.replace(/\"/g, '""') + '"';
+    } else {
+        return txt;
+    }
+}
+
+// Parse PostgreSQL table name (possibly quoted and with optional schema).+
+// Returns { schema: 'schema_name', table: 'table_name' }
+function parse_table_name(table) {
+
+    function split_as_quoted_parts(table_name) {
+      // parse table into 'parts' that may be quoted, each part
+      // in the parts array being an object { part: 'text', quoted: false/true }
+     var parts = [];
+     var splitted = table_name.split(/\"/);
+     for (var i=0; i<splitted.length; i++ ) {
+       if ( splitted[i] === '' ) {
+         if ( parts.length > 0 && i < splitted.length-1 ) {
+             i++;
+             parts[parts.length - 1].part += '"' + splitted[i];
+         }
+       }
+       else {
+         var is_quoted = (i > 0 && splitted[i-1] === '') ||
+                         (i < splitted.length - 1 && splitted[i+1] === '');
+             parts.push({ part: splitted[i], quoted: is_quoted });
+       }
+     }
+     return parts;
+    }
+
+    var parts = split_as_quoted_parts(table);
+
+    function split_single_part(part) {
+        var schema_part = null;
+        var table_part = null;
+        if ( part.quoted ) {
+            table_part = part.part;
+        } else {
+            var parts = part.part.split('.');
+            if ( parts.length === 1 ) {
+                schema_part = null;
+                table_part = parts[0];
+            } else if ( parts.length === 2 ) {
+                schema_part = parts[0];
+                table_part = parts[1];
+            } // else invalid table name
+        }
+        return {
+            schema: schema_part,
+            table:  table_part
+        };
+    }
+
+    function split_two_parts(part1, part2) {
+        var schema_part = null;
+        var table_part = null;
+        if ( part1.quoted && !part2.quoted ) {
+            if ( part2.part[0] === '.' ) {
+                schema_part = part1.part;
+                table_part = part2.part.slice(1);
+            } // else invalid table name (missing dot)
+        } else if ( !part1.quoted && part2.quoted ) {
+            if ( part1.part[part1.part.length - 1] === '.' ) {
+                schema_part = part1.part.slice(0, -1);
+                table_part = part2.part;
+            } // else invalid table name (missing dot)
+        } // else invalid table name (missing dot)
+        return {
+            schema: schema_part,
+            table:  table_part
+        };
+    }
+
+    if ( parts.length === 1 ) {
+        return split_single_part(parts[0]);
+    } else if ( parts.length === 2 ) {
+        return split_two_parts(parts[0], parts[1]);
+    } else if ( parts.length === 3 && parts[1].part === '.' ) {
+        return {
+            schema: parts[0].part,
+            table:  parts[2].part
+        };
+    } // else invalid table name
+}
+
+function table_identifier(parsed_name) {
+    if ( parsed_name && parsed_name.table ) {
+        if ( parsed_name.schema ) {
+            return quote_identifier_if_needed(parsed_name.schema) + '.' + quote_identifier_if_needed(parsed_name.table);
+        } else {
+            return quote_identifier_if_needed(parsed_name.table);
+        }
+    } else {
+        return null;
+    }
+}
+
+module.exports = {
+  parse: parse_table_name,
+  quote: quote_identifier_if_needed,
+  table_identifier: table_identifier
+};

test/acceptance/overviews_metadata.js

@@ -1,4 +1,3 @@
-var _ = require('underscore');
 var test_helper = require('../support/test_helper');
 
 var assert = require('../support/assert');
@@ -15,7 +14,7 @@ var step = require('step');
 var windshaft = require('windshaft');
 
 
-describe('overviews', function() {
+describe('overviews metadata', function() {
     // configure redis pool instance to use in tests
     var redisPool = new RedisPool(global.environment.redis);
 
@@ -87,20 +86,16 @@ describe('overviews', function() {
               assert.ifError(err);
               assert.deepEqual(non_overviews_layer, mapConfig._cfg.layers[1]);
               assert.equal(mapConfig._cfg.layers[0].type, 'cartodb');
-              assert.ok(mapConfig._cfg.layers[0].options.overviews);
-              assert.ok(mapConfig._cfg.layers[0].options.overviews.test_table_overviews);
-              assert.deepEqual(_.keys(mapConfig._cfg.layers[0].options.overviews), ['test_table_overviews']);
-              assert.equal(_.keys(mapConfig._cfg.layers[0].options.overviews.test_table_overviews).length, 2);
-              assert.ok(mapConfig._cfg.layers[0].options.overviews.test_table_overviews[1]);
-              assert.ok(mapConfig._cfg.layers[0].options.overviews.test_table_overviews[2]);
-              assert.equal(
-                  mapConfig._cfg.layers[0].options.overviews.test_table_overviews[1].table,
-                  '_vovw_1_test_table_overviews'
-              );
-              assert.equal(
-                  mapConfig._cfg.layers[0].options.overviews.test_table_overviews[2].table,
-                  '_vovw_2_test_table_overviews'
-              );
+              assert.ok(mapConfig._cfg.layers[0].options.query_rewrite_data);
+              var expected_data = {
+                overviews: {
+                  test_table_overviews: {
+                    1: { table: '_vovw_1_test_table_overviews' },
+                    2: { table: '_vovw_2_test_table_overviews' }
+                  }
+                }
+              };
+              assert.deepEqual(mapConfig._cfg.layers[0].options.query_rewrite_data, expected_data);
             });
 
             next(err);

test/acceptance/overviews_queries.js

@@ -0,0 +1,72 @@
+var testHelper = require('../support/test_helper');
+var assert = require('../support/assert');
+
+var cartodbServer = require('../../lib/cartodb/server');
+var ServerOptions = require('./ported/support/ported_server_options');
+var testClient = require('./ported/support/test_client');
+var BaseController = require('../../lib/cartodb/controllers/base');
+
+describe('overviews_queries', function() {
+
+    var server = cartodbServer(ServerOptions);
+    server.setMaxListeners(0);
+
+    var IMAGE_EQUALS_TOLERANCE_PER_MIL = 2;
+
+    var req2paramsFn;
+    before(function() {
+        req2paramsFn = BaseController.prototype.req2params;
+        BaseController.prototype.req2params = ServerOptions.req2params;
+    });
+
+    after(function() {
+        BaseController.prototype.req2params = req2paramsFn;
+
+        testHelper.rmdirRecursiveSync(global.environment.millstone.cache_basedir);
+    });
+
+    function imageCompareFn(fixture, done) {
+        return function(err, tile) {
+            if (err) {
+                return done(err);
+            }
+            assert.imageEqualsFile(tile.body, './test/fixtures/' + fixture, IMAGE_EQUALS_TOLERANCE_PER_MIL, done);
+        };
+    }
+
+    it("should not use overview for tables without overviews", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table'), 1, 0, 0,
+          imageCompareFn('test_table_1_0_0.png', done)
+      );
+    });
+
+    it("should not use overview for tables without overviews at z=2", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table'), 2, 1, 1,
+          imageCompareFn('test_table_2_1_1.png', done)
+      );
+    });
+
+    it("should not use overview for tables without overviews at z=2", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table'), 3, 3, 3,
+          imageCompareFn('test_table_3_3_3.png', done)
+      );
+    });
+
+    it("should use overview for zoom level 1", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table_overviews'), 1, 0, 0,
+          imageCompareFn('_vovw_1_test_table_1_0_0.png', done)
+      );
+    });
+
+    it("should use overview for zoom level 1", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table_overviews'), 2, 1, 1,
+          imageCompareFn('_vovw_2_test_table_2_1_1.png', done)
+      );
+    });
+
+    it("should not use overview for zoom level 3", function(done){
+      testClient.getTile(testClient.defaultTableMapConfig('test_table_overviews'), 3, 3, 3,
+          imageCompareFn('test_table_3_3_3.png', done)
+      );
+    });
+});

test/acceptance/ported/support/ported_server_options.js

@@ -2,6 +2,10 @@ var _ = require('underscore');
 var serverOptions = require('../../../../lib/cartodb/server_options');
 var LayergroupToken = require('../../../../lib/cartodb/models/layergroup_token');
 var mapnik = require('windshaft').mapnik;
+var OverviewsQueryRewriter = require('../../../../lib/cartodb/utils/overviews_query_rewriter');
+var overviewsQueryRewriter = new OverviewsQueryRewriter({
+  zoom_level: 'CDB_ZoomFromScale(!scale_denominator!)'
+});
 
 module.exports = _.extend({}, serverOptions, {
     base_url: '/database/:dbname/table/:table',
@@ -26,7 +30,8 @@ module.exports = _.extend({}, serverOptions, {
             limits: {
                 render: 0,
                 cacheOnTimeout: true
-            }
+            },
+            queryRewriter: overviewsQueryRewriter
         },
         http: {
             timeout: 5000,

test/integration/mapconfig_overviews_adapter.js

@@ -1,12 +1,11 @@
 require('../support/test_helper');
 
 var assert = require('assert');
-var _ = require('underscore');
 var RedisPool = require('redis-mpool');
 var cartodbRedis = require('cartodb-redis');
 var PgConnection = require(__dirname + '/../../lib/cartodb/backends/pg_connection');
 var PgQueryRunner = require('../../lib/cartodb/backends/pg_query_runner');
-var OverviewsApi = require('../../lib/cartodb/api/overviews_api');
+var OverviewsMetadataApi = require('../../lib/cartodb/api/overviews_metadata_api');
 var MapConfigOverviewsAdapter = require('../../lib/cartodb/models/mapconfig_overviews_adapter');
 
 // configure redis pool instance to use in tests
@@ -17,10 +16,10 @@ var redisPool = new RedisPool(global.environment.redis);
 var metadataBackend = cartodbRedis({pool: redisPool});
 var pgConnection = new PgConnection(metadataBackend);
 var pgQueryRunner = new PgQueryRunner(pgConnection);
-var overviewsApi = new OverviewsApi(pgQueryRunner);
+var overviewsMetadataApi = new OverviewsMetadataApi(pgQueryRunner);
 
 
-var mapConfigOverviewsAdapter = new MapConfigOverviewsAdapter(overviewsApi);
+var mapConfigOverviewsAdapter = new MapConfigOverviewsAdapter(overviewsMetadataApi);
 
 describe('MapConfigOverviewsAdapter', function() {
 
@@ -72,20 +71,16 @@ describe('MapConfigOverviewsAdapter', function() {
             assert.equal(layers[0].options.sql, sql);
             assert.equal(layers[0].options.cartocss, cartocss);
             assert.equal(layers[0].options.cartocss_version, cartocss_version);
-            assert.ok(layers[0].options.overviews);
-            assert.ok(layers[0].options.overviews.test_table_overviews);
-            assert.deepEqual(_.keys(layers[0].options.overviews), ['test_table_overviews']);
-            assert.equal(_.keys(layers[0].options.overviews.test_table_overviews).length, 2);
-            assert.ok(layers[0].options.overviews.test_table_overviews[1]);
-            assert.ok(layers[0].options.overviews.test_table_overviews[2]);
-            assert.equal(
-                layers[0].options.overviews.test_table_overviews[1].table,
-                '_vovw_1_test_table_overviews'
-              );
-            assert.equal(
-                layers[0].options.overviews.test_table_overviews[2].table,
-                '_vovw_2_test_table_overviews'
-              );
+            assert.ok(layers[0].options.query_rewrite_data);
+            var expected_data = {
+              overviews: {
+                test_table_overviews: {
+                  1: { table: '_vovw_1_test_table_overviews' },
+                  2: { table: '_vovw_2_test_table_overviews' }
+                }
+              }
+            };
+            assert.deepEqual(layers[0].options.query_rewrite_data, expected_data);
             done();
         });
     });

test/integration/overviews-metadata-api.js

@@ -8,12 +8,12 @@ var cartodbRedis = require('cartodb-redis');
 var PgConnection = require('../../lib/cartodb/backends/pg_connection');
 var PgQueryRunner = require('../../lib/cartodb/backends/pg_query_runner');
 var QueryTablesApi = require('../../lib/cartodb/api/query_tables_api');
-var OverviewsApi = require('../../lib/cartodb/api/overviews_api');
+var OverviewsMetadataApi = require('../../lib/cartodb/api/overviews_metadata_api');
 
 
-describe('OverviewsApi', function() {
+describe('OverviewsMetadataApi', function() {
 
-    var queryTablesApi, overviewsApi;
+    var queryTablesApi, overviewsMetadataApi;
 
     before(function() {
         var redisPool = new RedisPool(global.environment.redis);
@@ -21,12 +21,12 @@ describe('OverviewsApi', function() {
         var pgConnection = new PgConnection(metadataBackend);
         var pgQueryRunner = new PgQueryRunner(pgConnection);
         queryTablesApi = new QueryTablesApi(pgQueryRunner);
-        overviewsApi = new OverviewsApi(pgQueryRunner);
+        overviewsMetadataApi = new OverviewsMetadataApi(pgQueryRunner);
     });
 
     it('should return an empty relation for tables that have no overviews', function(done) {
         var query = 'select * from test_table';
-        overviewsApi.getOverviewsMetadata('localhost', query, function(err, result) {
+        overviewsMetadataApi.getOverviewsMetadata('localhost', query, function(err, result) {
             assert.ok(!err, err);
 
             assert.deepEqual(result, {});
@@ -37,7 +37,7 @@ describe('OverviewsApi', function() {
 
     it('should return overviews metadata', function(done) {
         var query = 'select * from test_table_overviews';
-        overviewsApi.getOverviewsMetadata('localhost', query, function(err, result) {
+        overviewsMetadataApi.getOverviewsMetadata('localhost', query, function(err, result) {
             assert.ok(!err, err);
 
             assert.deepEqual(result, {

test/support/sql/CDB_Overviews.sql

@@ -16,6 +16,31 @@ $$ LANGUAGE PLPGSQL;
 
 CREATE OR REPLACE FUNCTION CDB_ZoomFromScale(scaleDenominator numeric) RETURNS int AS $$
 BEGIN
-  RETURN 0;
+  CASE
+    WHEN scaleDenominator > 500000000 THEN RETURN 0;
+    WHEN scaleDenominator <= 500000000 AND scaleDenominator > 200000000 THEN RETURN 1;
+    WHEN scaleDenominator <= 200000000 AND scaleDenominator > 100000000 THEN RETURN 2;
+    WHEN scaleDenominator <= 100000000 AND scaleDenominator > 50000000 THEN RETURN 3;
+    WHEN scaleDenominator <= 50000000 AND scaleDenominator > 25000000 THEN RETURN 4;
+    WHEN scaleDenominator <= 25000000 AND scaleDenominator > 12500000 THEN RETURN 5;
+    WHEN scaleDenominator <= 12500000 AND scaleDenominator > 6500000 THEN RETURN 6;
+    WHEN scaleDenominator <= 6500000 AND scaleDenominator > 3000000 THEN RETURN 7;
+    WHEN scaleDenominator <= 3000000 AND scaleDenominator > 1500000 THEN RETURN 8;
+    WHEN scaleDenominator <= 1500000 AND scaleDenominator > 750000 THEN RETURN 9;
+    WHEN scaleDenominator <= 750000 AND scaleDenominator > 400000 THEN RETURN 10;
+    WHEN scaleDenominator <= 400000 AND scaleDenominator > 200000 THEN RETURN 11;
+    WHEN scaleDenominator <= 200000 AND scaleDenominator > 100000 THEN RETURN 12;
+    WHEN scaleDenominator <= 100000 AND scaleDenominator > 50000 THEN RETURN 13;
+    WHEN scaleDenominator <= 50000 AND scaleDenominator > 25000 THEN RETURN 14;
+    WHEN scaleDenominator <= 25000 AND scaleDenominator > 12500 THEN RETURN 15;
+    WHEN scaleDenominator <= 12500 AND scaleDenominator > 5000 THEN RETURN 16;
+    WHEN scaleDenominator <= 5000 AND scaleDenominator > 2500 THEN RETURN 17;
+    WHEN scaleDenominator <= 2500 AND scaleDenominator > 1500 THEN RETURN 18;
+    WHEN scaleDenominator <= 1500 AND scaleDenominator > 750 THEN RETURN 19;
+    WHEN scaleDenominator <= 750 AND scaleDenominator > 500 THEN RETURN 20;
+    WHEN scaleDenominator <= 500 AND scaleDenominator > 250 THEN RETURN 21;
+    WHEN scaleDenominator <= 250 AND scaleDenominator > 100 THEN RETURN 22;
+    WHEN scaleDenominator <= 100 THEN RETURN 23;
+  END CASE;
 END
 $$ LANGUAGE plpgsql IMMUTABLE;

test/support/sql/windshaft.test.sql

@@ -325,3 +325,10 @@ CREATE TABLE _vovw_2_test_table_overviews (
 
 GRANT ALL ON TABLE _vovw_2_test_table_overviews TO :TESTUSER;
 GRANT SELECT ON TABLE _vovw_2_test_table_overviews TO :PUBLICUSER;
+
+INSERT INTO _vovw_2_test_table_overviews VALUES
+('2011-09-21 14:02:21.358706', '2011-09-21 14:02:21.314252', 1, 'Hawai', 'Calle de Pérez Galdós 9, Madrid, Spain', '0101000020E610000000000000000020C00000000000004440', '0101000020110F000076491621312319C122D4663F1DCC5241'),
+('2011-09-21 14:02:21.358706', '2011-09-21 14:02:21.319101', 2, 'El Estocolmo', 'Calle de la Palma 72, Madrid, Spain', '0101000020E610000000000000009431C026043C75E7224340', '0101000020110F0000C4356B29423319C15DD1092DADCC5241');
+
+INSERT INTO _vovw_1_test_table_overviews VALUES
+('2011-09-21 14:02:21.358706', '2011-09-21 14:02:21.314252', 1, 'Hawai', 'Calle de Pérez Galdós 9, Madrid, Spain', '0101000020E610000000000000000020C00000000000004440', '0101000020110F000076491621312319C122D4663F1DCC5241');

test/unit/cartodb/overviews_query_rewriter.js

@@ -0,0 +1,429 @@
+require('../../support/test_helper');
+
+var assert           = require('assert');
+var OverviewsQueryRewriter = require('../../../lib/cartodb/utils/overviews_query_rewriter');
+var overviewsQueryRewriter = new OverviewsQueryRewriter({
+  zoom_level: 'ZoomLevel()'
+});
+
+function normalize_whitespace(txt) {
+  return txt.replace(/\s+/g, " ").trim();
+}
+
+// compare SQL statements ignoring whitespace
+function assertSameSql(sql1, sql2) {
+    assert.equal(normalize_whitespace(sql1), normalize_whitespace(sql2));
+}
+
+describe('Overviews query rewriter', function() {
+
+  it('does not alter queries if no overviews data is present', function(done){
+    var sql = "SELECT * FROM table1";
+    var overviews_sql = overviewsQueryRewriter.query(sql);
+    assert.equal(overviews_sql, sql);
+    overviews_sql = overviewsQueryRewriter.query(sql, {});
+    assert.equal(overviews_sql, sql);
+    overviews_sql = overviewsQueryRewriter.query(sql, { overviews: {} });
+    assert.equal(overviews_sql, sql);
+    done();
+  });
+
+
+  it('does not alter queries which don\'t use overviews', function(done){
+    var sql = "SELECT * FROM table1";
+    var data = {
+        overviews: {
+            table2: {
+              0: { table: 'table2_ov0' },
+              1: { table: 'table2_ov1' },
+              4: { table: 'table2_ov4' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+    done();
+  });
+
+  // jshint multistr:true
+
+  it('generates query with single overview layer for level 0', function(done){
+    var sql = "SELECT * FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              0: { table: 'table1_ov0' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov0, _vovw_scale WHERE _vovw_z = 0\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 0\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query with single overview layer for level >0', function(done){
+    var sql = "SELECT * FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query with multiple overview layers for all levels up to N', function(done){
+    var sql = "SELECT * FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              0: { table: 'table1_ov0' },
+              1: { table: 'table1_ov1' },
+              2: { table: 'table1_ov2' },
+              3: { table: 'table1_ov3' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov0, _vovw_scale WHERE _vovw_z = 0\
+            UNION ALL\
+            SELECT * FROM table1_ov1, _vovw_scale WHERE _vovw_z = 1\
+            UNION ALL\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z = 2\
+            UNION ALL\
+            SELECT * FROM table1_ov3, _vovw_scale WHERE _vovw_z = 3\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 3\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query with multiple overview layers for random levels', function(done){
+    var sql = "SELECT * FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              0: { table: 'table1_ov0' },
+              1: { table: 'table1_ov1' },
+              6: { table: 'table1_ov6' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov0, _vovw_scale WHERE _vovw_z = 0\
+            UNION ALL\
+            SELECT * FROM table1_ov1, _vovw_scale WHERE _vovw_z = 1\
+            UNION ALL\
+            SELECT * FROM table1_ov6, _vovw_scale WHERE _vovw_z > 1 AND _vovw_z <= 6\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 6\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query for a table with explicit schema', function(done){
+    var sql = "SELECT * FROM public.table1";
+    var data = {
+        overviews: {
+            'public.table1': {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM public.table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM public.table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query for a table with explicit schema in the overviews info', function(done){
+    var sql = "SELECT * FROM public.table1";
+    var data = {
+        overviews: {
+            'public.table1': {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM public.table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM public.table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query for a table that needs quoting with explicit schema', function(done){
+    var sql = "SELECT * FROM public.\"table 1\"";
+    var data = {
+        overviews: {
+            'public."table 1"': {
+              2: { table: '"table 1_ov2"' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          \"_vovw_table 1\" AS (\
+            SELECT * FROM public.\"table 1_ov2\", _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM public.\"table 1\", _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM \"_vovw_table 1\"\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query for a table with explicit schema that needs quoting', function(done){
+    var sql = "SELECT * FROM \"user-1\".table1";
+    var data = {
+        overviews: {
+            '"user-1".table1': {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM \"user-1\".table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM \"user-1\".table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+
+  it('generates query for a table with explicit schema both needing quoting', function(done){
+    var sql = "SELECT * FROM \"user-1\".\"table 1\"";
+    var data = {
+        overviews: {
+            '"user-1"."table 1"': {
+              2: { table: '"table 1_ov2"' }
+
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          \"_vovw_table 1\" AS (\
+            SELECT * FROM \"user-1\".\"table 1_ov2\", _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM \"user-1\".\"table 1\", _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT * FROM \"_vovw_table 1\"\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+
+  it('generates query using overviews for queries with selected columns', function(done){
+    var sql = "SELECT column1, column2, column3 FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT column1, column2, column3 FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query using overviews for queries with selected columns and all columns', function(done){
+    var sql = "SELECT table1.*, column1, column2, column3 FROM table1";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT _vovw_table1.*, column1, column2, column3 FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query using overviews for queries with a semicolon', function(done){
+    var sql = "SELECT table1.*, column1, column2, column3 FROM table1;";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT _vovw_table1.*, column1, column2, column3 FROM _vovw_table1;\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('generates query using overviews for queries with extra whitespace', function(done){
+    var sql = "  SELECT  table1.* , column1,column2,  column3 FROM  table1  ";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    var expected_sql = "\
+        WITH\
+          _vovw_scale AS ( SELECT ZoomLevel() AS _vovw_z ),\
+          _vovw_table1 AS (\
+            SELECT * FROM table1_ov2, _vovw_scale WHERE _vovw_z <= 2\
+            UNION ALL\
+            SELECT * FROM table1, _vovw_scale WHERE _vovw_z > 2\
+          )\
+        SELECT _vovw_table1.* , column1,column2, column3 FROM _vovw_table1\
+    ";
+    assertSameSql(overviews_sql, expected_sql);
+    done();
+  });
+
+  it('does not alter queries which have not the simple supported form', function(done){
+    var sql = "SELECT * FROM table1 WHERE column1='x'";
+    var data = {
+        overviews: {
+            table1: {
+              2: { table: 'table1_ov2' }
+            }
+        }
+    };
+    var overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "SELECT * FROM table1 JOIN table2 ON (table1.col1=table2.col1)";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "SELECT a+b AS c FROM table1";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "SELECT f(a) AS b FROM table1";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "SELECT * FROM table1 AS x";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "WITH a AS (1) SELECT * FROM table1";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "SELECT * FROM table1 WHERE a=1";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    sql = "\
+        SELECT table1.* FROM table1 \
+               JOIN areas ON ST_Intersects(table1.the_geom, areas.the_geom) \
+               WHERE areas.name='A' \
+          ";
+    overviews_sql = overviewsQueryRewriter.query(sql, data);
+    assert.equal(overviews_sql, sql);
+
+    done();
+  });
+
+
+});

test/unit/cartodb/table_name_parser.js

@@ -0,0 +1,60 @@
+require('../../support/test_helper');
+
+var assert        = require('assert');
+var TableNameParser     = require('../../../lib/cartodb/utils/table_name_parser');
+
+describe('TableNameParser', function() {
+
+  it('parses table names with scheme and quotes', function(done){
+
+      var test_cases = [
+        ['xyz',             { schema: null,       table: 'xyz' }],
+        ['"xyz"',           { schema: null,       table: 'xyz' }],
+        ['"xy z"',          { schema: null,       table: 'xy z' }],
+        ['"xy.z"',          { schema: null,       table: 'xy.z' }],
+        ['"x.y.z"',         { schema: null,       table: 'x.y.z' }],
+        ['abc.xyz',         { schema: 'abc',      table: 'xyz' }],
+        ['"abc".xyz',       { schema: 'abc',      table: 'xyz' }],
+        ['abc."xyz"',       { schema: 'abc',      table: 'xyz' }],
+        ['"abc"."xyz"',     { schema: 'abc',      table: 'xyz' }],
+        ['"a bc"."x yz"',   { schema: 'a bc',     table: 'x yz' }],
+        ['"a bc".xyz',      { schema: 'a bc',     table: 'xyz' }],
+        ['"a.bc".xyz',      { schema: 'a.bc',     table: 'xyz' }],
+        ['"a.b.c".xyz',     { schema: 'a.b.c',    table: 'xyz' }],
+        ['"a.b.c.".xyz',    { schema: 'a.b.c.',   table: 'xyz' }],
+        ['"a""bc".xyz',     { schema: 'a"bc',     table: 'xyz' }],
+        ['"a""bc"."x""yz"', { schema: 'a"bc',     table: 'x"yz' }],
+      ];
+
+      test_cases.forEach(function(test_case) {
+          var table_name = test_case[0];
+          var expected_result = test_case[1];
+          var result = TableNameParser.parse(table_name);
+          assert.deepEqual(result, expected_result);
+      });
+      done();
+  });
+
+  it('quotes identifiers that need quoting', function(done){
+      assert.equal(TableNameParser.quote('x yz'), '"x yz"');
+      assert.equal(TableNameParser.quote('x-yz'), '"x-yz"');
+      assert.equal(TableNameParser.quote('x.yz'), '"x.yz"');
+      done();
+  });
+
+  it('doubles quotes', function(done){
+      assert.equal(TableNameParser.quote('x"yz'), '"x""yz"');
+      assert.equal(TableNameParser.quote('x"y"z'), '"x""y""z"');
+      assert.equal(TableNameParser.quote('x""y"z'), '"x""""y""z"');
+      assert.equal(TableNameParser.quote('x "yz'), '"x ""yz"');
+      assert.equal(TableNameParser.quote('x"y-y"z'), '"x""y-y""z"');
+      done();
+  });
+
+  it('does not quote identifiers that don\'t need to be quoted', function(done){
+      assert.equal(TableNameParser.quote('xyz'), 'xyz');
+      assert.equal(TableNameParser.quote('x_z123'), 'x_z123');
+      done();
+  });
+
+});