Refactor metadata queries execution
Also fixed bug where sampling query generation needed results of count queries
This commit is contained in:
Javier Goizueta
@@ -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;
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user