diff --git a/lib/cartodb/backends/layer-stats/mapnik-layer-stats.js b/lib/cartodb/backends/layer-stats/mapnik-layer-stats.js index d5806960..047229af 100644 --- a/lib/cartodb/backends/layer-stats/mapnik-layer-stats.js +++ b/lib/cartodb/backends/layer-stats/mapnik-layer-stats.js @@ -1,4 +1,5 @@ var queryUtils = require('../../utils/query-utils'); +const PhasedExecution = require('../../utils/phased-execution'); const AggregationMapConfig = require('../../models/aggregation/aggregation-mapconfig'); function MapnikLayerStats () { @@ -27,154 +28,156 @@ function queryPromise(dbConnection, query, callback) { }); } +function columnAggregations(field) { + if (field.type === 'number') { + return ['min', 'max', 'avg', 'sum']; + } + if (field.type === 'date') { // TODO other types too? + return ['min', 'max']; + } +} + +function firstPhaseQueries(queries, ctx) { + if (queries.results.estimatedFeatureCount === undefined) { + queries.task( + queryPromise(ctx.dbConnection, queryUtils.getQueryRowEstimation(ctx.query), function(err, res) { + if (err) { + // at least for debugging we should err + queries.results.estimatedFeatureCount = -1; + return null; + } else { + // We decided that the relation is 1 row == 1 feature + queries.results.estimatedFeatureCount = res.rows[0].rows; + return null; + } + }) + ); + } + + if (ctx.metaOptions.featureCount) { + // TODO: if ctx.metaOptions.columnStats we can combine this with column stats query + queries.task( + queryPromise( + queryUtils.getQueryActualRowCount(ctx.rawQuery), + function(err, res) { + if (err) { + queries.results.featureCount = -1; + } else { + queries.results.featureCount = res.rows[0].rows; + } + return err; + } + ) + ); + } + + if (ctx.metaOptions.geometryType && queries.results.geometryType === undefined) { + const geometryColumn = AggregationMapConfig.getAggregationGeometryColumn(); + queries.task( + queryPromise(queryUtils.getQueryGeometryType(ctx.rawQuery, geometryColumn), function(err, res) { + if (!err) { + queries.results.geometryType = res.geom_type; + } + return err; + }) + ); + } + + if (ctx.metaOptions.columns || ctx.metaOptions.columnStats) { + queries.task( + // TODO: note we have getLayerColumns in aggregation mapconfig. + // and also getLayerAggregationColumns which either uses getLayerColumns or derives columns from parameters + queryPromise(queryUtils.getQueryLimited(ctx.rawQuery, 0), function(err, res) { + if (!err) { + queries.results.columns = res.fields; + } + return err; + }) + ); + } +} + +function secondPhaseQueries(queries, ctx) { + if (ctx.metaOptions.sample) { + const numRows = queries.results.featureCount === undefined ? + queries.results.estimatedFeatureCount : + queries.results.featureCount; + const sampleProb = Math.min(ctx.metaOptions.sample / numRows, 1); + queries.task( + queryPromise( + queryUtils.getQuerySample(ctx.rawQuery, sampleProb), + function(err, res) { + if (err) { + queries.results.sample = []; + } else { + queries.results.sample = res.rows; + } + return err; + } + ) + ); + } + + if (ctx.metaOptions.columnStats) { + let aggr = []; + Object.keys(queries.results.columns).forEach(name => { + aggr = aggr.concat(columnAggregations(queries.results.columns[name]) + .map(fn => `${fn}(${name}) AS ${name}_${fn}`)); + if (queries.results.columns[name].type === 'string') { + const topN = ctx.metaOptions.columnStats.topCategories || 1024; + // TODO: ctx.metaOptions.columnStats.maxCategories + // => use PG stats to dismiss columns with more distinct values + queries.task( + queryPromise(queryUtils.getQueryTopCategories(ctx.rawQuery, name, topN), function(err, res){ + if (!err) { + queries.results.columns[name].categories = res.rows; + } + return err; + }) + ); + } + }); + queries.task( + queryPromise(`SELECT ${aggr.join(',')} FROM (${ctx.rawQuery})`, function(err, res){ + if (!err) { + Object.keys(queries.results.columns).forEach(name => { + columnAggregations(queries.results.columns[name]).forEach(fn => { + queries.results.columns[name][fn] = res.rows[0][`${name}_${fn}`]; + }); + }); + } + return err; + }) + ); + } + +} + MapnikLayerStats.prototype.getStats = function (layer, dbConnection, callback) { - let query = layer.options.sql; - let rawQuery = layer.options.sql_raw ? layer.options.sql_raw : layer.options.sql; - let metaOptions = layer.options.metadata || {}; + let context = { + dbConnection, + query: layer.options.sql, + rawQuery: layer.options.sql_raw ? layer.options.sql_raw : layer.options.sql, + metaOptions: layer.options.metadata || {} + }; - let stats = {}; + let queries = new PhasedExecution(); // TODO: could save some queries if queryUtils.getAggregationMetadata() has been used and kept somewhere - // we would set stats.estimatedFeatureCount and stats.geometryType (if metaOptions.geometryType) from it. + // we would set queries.results.estimatedFeatureCount and queries.results.geometryType + // (if metaOptions.geometryType) from it. // We'll add promises for queries to be executed to the next two lists; // the queries in statQueries2 will be executed after all of statQueries are completed, // so any results from them can be used. // Query promises will store results in the shared stats object. - let statQueries = [], statQueries2 = []; - if (stats.estimatedFeatureCount === undefined) { - statQueries.push( - queryPromise(dbConnection, queryUtils.getQueryRowEstimation(query), function(err, res) { - if (err) { - // at least for debugging we should err - stats.estimatedFeatureCount = -1; - return null; - } else { - // We decided that the relation is 1 row == 1 feature - stats.estimatedFeatureCount = res.rows[0].rows; - return null; - } - }) - ); - } - - if (metaOptions.featureCount) { - // TODO: if metaOptions.columnStats we can combine this with column stats query - statQueries.push( - queryPromise( - queryUtils.getQueryActualRowCount(rawQuery), - function(err, res) { - if (err) { - stats.featureCount = -1; - } else { - stats.featureCount = res.rows[0].rows; - } - return err; - } - ) - ); - } - - if (metaOptions.sample) { - const numRows = stats.featureCount === undefined ? stats.estimatedFeatureCount : stats.featureCount; - const sampleProb = Math.min(metaOptions.sample / numRows, 1); - statQueries2.push( - queryPromise( - queryUtils.getQuerySample(rawQuery, sampleProb), - function(err, res) { - if (err) { - stats.sample = []; - } else { - stats.sample = res.rows; - } - return err; - } - ) - ); - } - - if (metaOptions.geometryType && stats.geometryType === undefined) { - const geometryColumn = AggregationMapConfig.getAggregationGeometryColumn(); - statQueries.push( - queryPromise(queryUtils.getQueryGeometryType(rawQuery, geometryColumn), function(err, res) { - if (!err) { - stats.geometryType = res.geom_type; - } - return err; - }) - ); - } - - function columnAggregations(field) { - if (field.type === 'number') { - return ['min', 'max', 'avg', 'sum']; - } - if (field.type === 'date') { // TODO other types too? - return ['min', 'max']; - } - } - - if (metaOptions.columns || metaOptions.columnStats) { - statQueries.push( - // TODO: note we have getLayerColumns in aggregation mapconfig. - // and also getLayerAggregationColumns which either uses getLayerColumns or derives columns from parameters - queryPromise(queryUtils.getQueryLimited(rawQuery, 0), function(err, res) { - if (!err) { - stats.columns = res.fields; - if (metaOptions.columnStats) { - let aggr = []; - Object.keys(stats.columns).forEach(name => { - aggr = aggr.concat(columnAggregations(stats.columns[name]) - .map(fn => `${fn}(${name}) AS ${name}_${fn}`)); - if (stats.columns[name].type === 'string') { - statQueries2.push( - queryPromise(topQuery(rawQuery, name, N), function(err, res){ - if (!err) { - const topN = metaOptions.columnStats.topCategories || 1024; - // TODO: metaOptions.columnStats.maxCategories => use PG stats to dismiss columns with more distinct values - statQueries2.push( - queryPromise( - queryUtils.getQueryTopCategories(rawQuery, topN), - function(err, res) { - if (!err) { - stats.columns[name].categories = res.rows; - } - return err; - } - ) - ); - - } - return err; - }) - ); - } - }) - statQueries2.push( - queryPromise(`SELECT ${aggr.join(',')} FROM (${rawQuery})`, function(err, res){ - if (!err) { - Object.keys(stats.columns).forEach(name => { - columnAggregations(stats.columns[name]).forEach(fn => { - stats.columns[name][fn] = res.rows[0][`${name}_${fn}`] - }); - }); - } - return err; - }) - ); - } - } - return err; - }) - ); - - } - - Promise.all(statQueries).then( () => { - Promise.all(statQueries2).then( () => callback(null, stats) ).catch( err => callback(err) ); - }).catch( err => callback(err) ); + // Queries will be executed in two phases, with results from the first phase needed + // to define the queries of the second phase + queries.phase(() => firstPhaseQueries(queries, context)); + queries.phase(() => secondPhaseQueries(queries, context)); + queries.run(callback); }; module.exports = MapnikLayerStats; diff --git a/lib/cartodb/utils/phased-execution.js b/lib/cartodb/utils/phased-execution.js new file mode 100644 index 00000000..ec1de9b1 --- /dev/null +++ b/lib/cartodb/utils/phased-execution.js @@ -0,0 +1,91 @@ +/** + * PhasedExecution handles the execution of async tasks (via Promises) + * which have dependencies between them in a simplified manner. + * Instead of using the complete task dependency graph, tasks + * are organized into execution phases. So that tasks from a latter + * phase will be initialized after tasks from previous phases have + * finished. + * + * All tasks place their results in a shared object to make them + * available to tasks of latter phases. + * + * Each phase is defined by a function that defines its tasks. + * + * Example: + * + * let p = new PhasedExecution(); + * // Define first phase with tasks 1 & 2 + * p.phase(() => { + * p.results.phase1 = 1 + * p.task(new Promise((resolve) => { + * setTimeout( () => { + * console.log('At 1:', p.results); + * p.results.task1 = 100; + * resolve(); + * }, 400); + * })); + * p.task(new Promise((resolve) => { + * setTimeout( () => { + * console.log('At 2:', p.results); + * p.results.task2 = 200; + * resolve(); + * }, 100); + * })); + * }); + * // Define second phase with tasks 3 & 4 + * p.phase(() => { + * p.results.phase2 = 2 + * p.task(new Promise((resolve) => { + * setTimeout( () => { + * console.log('At 3:', p.results); + * p.results.task3 = 300; + * resolve(); + * }, 50); + * })); + * p.task(new Promise((resolve) => { + * setTimeout( () => { + * console.log('At 4:', p.results); + * p.results.task4 = 400; + * resolve(); + * }, 100); + * })); + * }); + * // Define third phase with task 5 + * p.phase(() => { + * p.results.phase3 = 3 + * p.task(new Promise((resolve) => { + * setTimeout( () => { + * console.log('At 5:', p.results); + * p.results.task5 = 500; + * resolve(); + * }, 50); + * })); + * }); + * // Execute all tasks + * p.run(() => { + * console.log("RESULTS:", p.results); + * }); + */ +module.exports = class PhasedExecution { + constructor() { + this.results = {}; + this.phases = []; + } + phase(phasegenerator) { + this.phases.push(phasegenerator); + } + task(promise) { + this.tasks.push(promise); + } + run(callback) { + this.tasks = []; + let phase = this.phases.shift(); + if (phase) { + phase(this); + return Promise.all(this.tasks) + .then(this.run()) + .then(() => { if (callback) { callback(this.results); } }); + } + } +}; +// TODO: error handling