Fix estimated row count with aggregations
All stats are computed now pre-aggregation Code to help compute post-aggregation stats remains for testing
This commit is contained in:
Javier Goizueta
@@ -1,6 +1,24 @@
|
||||
var queryUtils = require('../../utils/query-utils');
|
||||
const PhasedExecution = require('../../utils/phased-execution');
|
||||
const AggregationMapConfig = require('../../models/aggregation/aggregation-mapconfig');
|
||||
var SubstitutionTokens = require('../../utils/substitution-tokens');
|
||||
|
||||
// Instantiate a query with tokens for a given zoom level
|
||||
function queryForZoom(sql, zoom) {
|
||||
const tileRes = 256;
|
||||
const wmSize = 6378137.0*2*Math.PI;
|
||||
const nTiles = Math.pow(2, zoom);
|
||||
const tileSize = wmSize / nTiles;
|
||||
const resolution = tileSize / tileRes;
|
||||
const scaleDenominator = resolution / 0.00028;
|
||||
const x0 = -wmSize/2, y0 = -wmSize/2;
|
||||
return SubstitutionTokens.replace(sql, {
|
||||
bbox: `ST_MakeEnvelope(${x0}, ${y0}, ${x0 + tileSize}, ${y0 + tileSize})`,
|
||||
scale_denominator: scaleDenominator,
|
||||
pixel_width: resolution,
|
||||
pixel_height: resolution
|
||||
});
|
||||
}
|
||||
|
||||
function MapnikLayerStats () {
|
||||
this._types = {
|
||||
@@ -9,8 +27,6 @@ function MapnikLayerStats () {
|
||||
};
|
||||
}
|
||||
|
||||
// TODO: estimatedFeatureCount is post-aggregation; the rest is pre; distinguish and complement
|
||||
|
||||
MapnikLayerStats.prototype.is = function (type) {
|
||||
return this._types[type] ? this._types[type] : false;
|
||||
};
|
||||
@@ -40,106 +56,109 @@ function columnAggregations(field) {
|
||||
return [];
|
||||
}
|
||||
|
||||
/* Helper to add a task to the queries PhasedExecution
|
||||
* type can be either 'pre' (for pre-aggregation metadata) or 'post'
|
||||
* zoom is used only for post-aggregation metadata
|
||||
* query is a function that generates a metadata query from a data query
|
||||
* assign is a function to assign the results of the metadata query
|
||||
* if a assignDefault function is present, it will be used in case of error
|
||||
* during the query execution and any errors will be ignored
|
||||
*/
|
||||
function addStat(queries, ctx, type, zoom, query, assign, assignDefault=null) {
|
||||
let sql;
|
||||
if (type === 'pre') {
|
||||
sql = ctx. preQuery;
|
||||
}
|
||||
else {
|
||||
sql = queryForZoom(ctx.aggrQuery, zoom);
|
||||
}
|
||||
sql = query(sql);
|
||||
queries.task(
|
||||
queryPromise(
|
||||
ctx.dbConnection,
|
||||
sql,
|
||||
(err, res) => {
|
||||
if (!err) {
|
||||
assign(res);
|
||||
}
|
||||
else if (assignDefault !== null) {
|
||||
assignDefault();
|
||||
return null;
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
);
|
||||
}
|
||||
function firstPhaseQueries(queries, ctx) {
|
||||
// estimatedFeatureCount
|
||||
if (queries.results.estimatedFeatureCount === undefined) {
|
||||
queries.task(
|
||||
queryPromise(ctx.dbConnection, queryUtils.getQueryRowEstimation(ctx.aggrQuery), 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;
|
||||
}
|
||||
})
|
||||
// This is always computed; a default value of -1 is used in case of error
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
queryUtils.getQueryRowEstimation,
|
||||
res => queries.results.estimatedFeatureCount = res.rows[0].rows,
|
||||
() => queries.results.estimatedFeatureCount = -1
|
||||
);
|
||||
}
|
||||
|
||||
// featureCount
|
||||
if (ctx.metaOptions.featureCount) {
|
||||
// TODO: pre/aggr
|
||||
// TODO: for pre, use ctx.aggrMeta.pre_aggregation_count
|
||||
// TODO: if ctx.metaOptions.columnStats we can combine this with column stats query
|
||||
queries.task(
|
||||
queryPromise(
|
||||
ctx.dbConnection,
|
||||
queryUtils.getQueryActualRowCount(ctx.preQuery),
|
||||
(err, res) => {
|
||||
if (err) {
|
||||
queries.results.featureCount = -1;
|
||||
} else {
|
||||
queries.results.featureCount = res.rows[0].rows;
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
queryUtils.getQueryActualRowCount,
|
||||
res => queries.results.featureCount = res.rows[0].rows
|
||||
);
|
||||
}
|
||||
|
||||
// geometryType
|
||||
if (ctx.metaOptions.geometryType && queries.results.geometryType === undefined) {
|
||||
// TODO: pre/aggr
|
||||
// TODO: for pre, use ctx.aggrMeta.geometry_type
|
||||
const geometryColumn = AggregationMapConfig.getAggregationGeometryColumn();
|
||||
queries.task(
|
||||
queryPromise(
|
||||
ctx.dbConnection,
|
||||
queryUtils.getQueryGeometryType(ctx.preQuery, geometryColumn),
|
||||
(err, res) => {
|
||||
if (!err) {
|
||||
queries.results.geometryType = res.rows[0].geom_type;
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
sql => queryUtils.getQueryGeometryType(sql, geometryColumn),
|
||||
res => queries.results.geometryType = res.rows[0].geom_type
|
||||
);
|
||||
}
|
||||
|
||||
// columns (names & types)
|
||||
if (ctx.metaOptions.columns || ctx.metaOptions.columnStats) {
|
||||
// TODO: pre/aggr
|
||||
// TODO: for aggr, use layer.options.columns (will need to pass in ctx)
|
||||
// note: post-aggregation columns are in layer.options.columns when aggregation is present
|
||||
queries.task(
|
||||
// TODO: note we have getLayerColumns in aggregation mapconfig.
|
||||
// and also getLayerAggregationColumns which either uses getLayerColumns or derives columns from parameters
|
||||
queryPromise(
|
||||
ctx.dbConnection,
|
||||
queryUtils.getQueryLimited(ctx.preQuery, 0),
|
||||
(err, res) => {
|
||||
if (!err) {
|
||||
queries.results.columns = formatResultFields(ctx.dbConnection, res.fields);
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
sql => queryUtils.getQueryLimited(sql, 0),
|
||||
res => queries.results.columns = formatResultFields(ctx.dbConnection, res.fields)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function secondPhaseQueries(queries, ctx) {
|
||||
// sample
|
||||
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(
|
||||
ctx.dbConnection,
|
||||
queryUtils.getQuerySample(ctx.preQuery, sampleProb),
|
||||
(err, res) => {
|
||||
if (err) {
|
||||
queries.results.sample = [];
|
||||
} else {
|
||||
queries.results.sample = res.rows;
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
sql => queryUtils.getQuerySample(sql, sampleProb),
|
||||
res => queries.results.sample = res.rows
|
||||
);
|
||||
}
|
||||
|
||||
// columnStats
|
||||
if (ctx.metaOptions.columnStats) {
|
||||
// TODO: pre/aggr
|
||||
let aggr = [];
|
||||
Object.keys(queries.results.columns).forEach(name => {
|
||||
aggr = aggr.concat(
|
||||
@@ -150,35 +169,27 @@ function secondPhaseQueries(queries, ctx) {
|
||||
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(
|
||||
ctx.dbConnection,
|
||||
queryUtils.getQueryTopCategories(ctx.preQuery, name, topN),
|
||||
(err, res) => {
|
||||
if (!err) {
|
||||
queries.results.columns[name].categories = res.rows;
|
||||
}
|
||||
return err;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
sql => queryUtils.getQueryTopCategories(sql, name, topN),
|
||||
res => queries.results.columns[name].categories = res.rows
|
||||
);
|
||||
}
|
||||
});
|
||||
queries.task(
|
||||
queryPromise(
|
||||
ctx.dbConnection,
|
||||
`SELECT ${aggr.join(',')} FROM (${ctx.preQuery}) AS __cdb_query`,
|
||||
(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;
|
||||
}
|
||||
)
|
||||
addStat(
|
||||
queries,
|
||||
ctx,
|
||||
'pre', 0,
|
||||
sql => `SELECT ${aggr.join(',')} FROM (${sql}) AS __cdb_query`,
|
||||
res => {
|
||||
Object.keys(queries.results.columns).forEach(name => {
|
||||
columnAggregations(queries.results.columns[name]).forEach(fn => {
|
||||
queries.results.columns[name][fn] = res.rows[0][`${name}_${fn}`];
|
||||
});
|
||||
});
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
@@ -209,6 +220,8 @@ function fieldType(cname) {
|
||||
return tname;
|
||||
}
|
||||
|
||||
// columns are returned as an object { columnName1: { type1: ...}, ..}
|
||||
// for consistency with SQL API
|
||||
function formatResultFields(dbConnection, flds) {
|
||||
flds = flds || [];
|
||||
var nfields = {};
|
||||
@@ -229,7 +242,9 @@ function formatResultFields(dbConnection, flds) {
|
||||
MapnikLayerStats.prototype.getStats =
|
||||
function (layer, dbConnection, callback) {
|
||||
let aggrQuery = layer.options.sql_raw || layer.options.sql;
|
||||
let preQuery = layer.options.aggregation_metadata ? layer.options.aggregation_metadata.pre_aggregation_sql : aggrQuery;
|
||||
let preQuery = layer.options.aggregation_metadata ?
|
||||
layer.options.aggregation_metadata.pre_aggregation_sql :
|
||||
aggrQuery;
|
||||
|
||||
let context = {
|
||||
dbConnection,
|
||||
|
||||
Reference in New Issue
Block a user