From e93385e306a889075b1f74506b938e390c8f2d55 Mon Sep 17 00:00:00 2001 From: Allen Short Date: Tue, 27 Mar 2018 20:58:26 +0000 Subject: [PATCH] Aggregate query results (re #35) (#339) --- .../components/queries/schedule-dialog.html | 3 + .../app/components/queries/schedule-dialog.js | 13 ++- client/app/pages/alerts-list/index.js | 1 - client/app/pages/queries/view.js | 1 + client/app/services/query-result.js | 10 +++ client/app/services/query.js | 6 +- migrations/versions/9d7678c47452_.py | 34 ++++++++ redash/handlers/api.py | 3 +- redash/handlers/queries.py | 3 + redash/handlers/query_results.py | 29 ++++++- redash/models.py | 53 +++++++++++- redash/serializers.py | 1 + redash/tasks/queries.py | 1 + tests/factories.py | 7 +- tests/handlers/test_queries.py | 80 +++++++++++++++++++ tests/test_models.py | 64 +++++++++++++-- 16 files changed, 294 insertions(+), 15 deletions(-) create mode 100644 migrations/versions/9d7678c47452_.py diff --git a/client/app/components/queries/schedule-dialog.html b/client/app/components/queries/schedule-dialog.html index f9344238a1..aca492cdfe 100644 --- a/client/app/components/queries/schedule-dialog.html +++ b/client/app/components/queries/schedule-dialog.html @@ -19,4 +19,7 @@

Refresh Schedule

Stop scheduling at date/time (format yyyy-MM-ddTHH:mm:ss, like 2016-12-28T14:57:00): + diff --git a/client/app/components/queries/schedule-dialog.js b/client/app/components/queries/schedule-dialog.js index db6ebe0320..41c29e031c 100644 --- a/client/app/components/queries/schedule-dialog.js +++ b/client/app/components/queries/schedule-dialog.js @@ -114,11 +114,21 @@ function scheduleUntil() { }; } +function scheduleKeepResults() { + return { + restrict: 'E', + scope: { + query: '=', + saveQuery: '=', + }, + template: '', + }; +} + const ScheduleForm = { controller() { this.query = this.resolve.query; this.saveQuery = this.resolve.saveQuery; - if (this.query.hasDailySchedule()) { this.refreshType = 'daily'; } else { @@ -137,5 +147,6 @@ export default function init(ngModule) { ngModule.directive('queryTimePicker', queryTimePicker); ngModule.directive('queryRefreshSelect', queryRefreshSelect); ngModule.directive('scheduleUntil', scheduleUntil); + ngModule.directive('scheduleKeepResults', scheduleKeepResults); ngModule.component('scheduleDialog', ScheduleForm); } diff --git a/client/app/pages/alerts-list/index.js b/client/app/pages/alerts-list/index.js index 082ec203c1..19869bcc5b 100644 --- a/client/app/pages/alerts-list/index.js +++ b/client/app/pages/alerts-list/index.js @@ -9,7 +9,6 @@ const stateClass = { class AlertsListCtrl { constructor(Alert) { - this.showEmptyState = false; this.showList = false; diff --git a/client/app/pages/queries/view.js b/client/app/pages/queries/view.js index 3336e21d56..6d864c8c02 100644 --- a/client/app/pages/queries/view.js +++ b/client/app/pages/queries/view.js @@ -205,6 +205,7 @@ function QueryViewCtrl( } else { request = pick($scope.query, [ 'schedule', + 'schedule_resultset_size', 'query', 'id', 'description', diff --git a/client/app/services/query-result.js b/client/app/services/query-result.js index 8e31a012a8..a25ac78cc0 100644 --- a/client/app/services/query-result.js +++ b/client/app/services/query-result.js @@ -54,6 +54,7 @@ function addPointToSeries(point, seriesCollection, seriesName) { function QueryResultService($resource, $timeout, $q, QueryResultError) { const QueryResultResource = $resource('api/query_results/:id', { id: '@id' }, { post: { method: 'POST' } }); + const QueryResultSetResource = $resource('api/queries/:id/resultset', { id: '@id' }); const Job = $resource('api/jobs/:id', { id: '@id' }); const statuses = { 1: 'waiting', @@ -452,6 +453,15 @@ function QueryResultService($resource, $timeout, $q, QueryResultError) { return queryResult; } + static getResultSet(queryId) { + const queryResult = new QueryResult(); + + QueryResultSetResource.get({ id: queryId }, (response) => { + queryResult.update(response); + }); + + return queryResult; + } loadResult(tryCount) { this.isLoadingResult = true; QueryResultResource.get( diff --git a/client/app/services/query.js b/client/app/services/query.js index ec64d34f61..76477ee88b 100644 --- a/client/app/services/query.js +++ b/client/app/services/query.js @@ -449,7 +449,11 @@ function QueryResource( this.latest_query_data_id = null; } - if (this.latest_query_data && maxAge !== 0) { + if (this.schedule_resultset_size) { + if (!this.queryResult) { + this.queryResult = QueryResult.getResultSet(this.id); + } + } else if (this.latest_query_data && maxAge !== 0) { if (!this.queryResult) { this.queryResult = new QueryResult({ query_result: this.latest_query_data, diff --git a/migrations/versions/9d7678c47452_.py b/migrations/versions/9d7678c47452_.py new file mode 100644 index 0000000000..d351153c87 --- /dev/null +++ b/migrations/versions/9d7678c47452_.py @@ -0,0 +1,34 @@ +"""Incremental query results aggregation + +Revision ID: 9d7678c47452 +Revises: 15041b7085fe +Create Date: 2018-03-08 04:36:12.802199 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '9d7678c47452' +down_revision = '15041b7085fe' +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table('query_resultsets', + sa.Column('query_id', sa.Integer(), nullable=False), + sa.Column('result_id', sa.Integer(), nullable=False), + sa.ForeignKeyConstraint(['query_id'], ['queries.id'], ), + sa.ForeignKeyConstraint(['result_id'], ['query_results.id'], ), + sa.PrimaryKeyConstraint('query_id', 'result_id') + ) + op.add_column(u'queries', sa.Column('schedule_resultset_size', sa.Integer(), nullable=True)) +1 + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column(u'queries', 'schedule_resultset_size') + op.drop_table('query_resultsets') + # ### end Alembic commands ### diff --git a/redash/handlers/api.py b/redash/handlers/api.py index fd6fe5fd15..5e7eb483f9 100644 --- a/redash/handlers/api.py +++ b/redash/handlers/api.py @@ -10,7 +10,7 @@ from redash.handlers.data_sources import DataSourceTypeListResource, DataSourceListResource, DataSourceSchemaResource, DataSourceResource, DataSourcePauseResource, DataSourceTestResource, DataSourceVersionResource from redash.handlers.events import EventsResource from redash.handlers.queries import QueryForkResource, QueryRefreshResource, QueryListResource, QueryRecentResource, QuerySearchResource, QueryResource, MyQueriesResource, QueryVersionListResource, ChangeResource -from redash.handlers.query_results import QueryResultListResource, QueryResultResource, JobResource +from redash.handlers.query_results import QueryResultListResource, QueryResultResource, JobResource, QueryResultSetResource from redash.handlers.users import UserResource, UserListResource, UserInviteResource, UserResetPasswordResource, UserDisableResource from redash.handlers.visualizations import VisualizationListResource from redash.handlers.visualizations import VisualizationResource @@ -85,6 +85,7 @@ def json_representation(data, code, headers=None): api.add_org_resource(QueryRefreshResource, '/api/queries//refresh', endpoint='query_refresh') api.add_org_resource(QueryResource, '/api/queries/', endpoint='query') api.add_org_resource(QueryForkResource, '/api/queries//fork', endpoint='query_fork') +api.add_org_resource(QueryResultSetResource, '/api/queries//resultset', endpoint='query_aggregate_results') api.add_org_resource(QueryVersionListResource, '/api/queries//version', endpoint='query_versions') api.add_org_resource(ChangeResource, '/api/changes/', endpoint='changes') diff --git a/redash/handlers/queries.py b/redash/handlers/queries.py index cb344ba7f5..86b24b7296 100644 --- a/redash/handlers/queries.py +++ b/redash/handlers/queries.py @@ -109,6 +109,7 @@ def post(self): : 0: + q.query_results.append(query_result) query_ids = [q.id for q in queries] logging.info("Updated %s queries with result (%s).", len(query_ids), query_hash) @@ -871,6 +875,7 @@ class Query(ChangeTrackingMixin, TimestampMixin, BelongsToOrgMixin, db.Model): data_source = db.relationship(DataSource, backref='queries') latest_query_data_id = Column(db.Integer, db.ForeignKey("query_results.id"), nullable=True) latest_query_data = db.relationship(QueryResult) + query_results = db.relationship("QueryResult", secondary="query_resultsets") name = Column(db.String(255)) description = Column(db.String(4096), nullable=True) query_text = Column("query", db.Text) @@ -886,6 +891,7 @@ class Query(ChangeTrackingMixin, TimestampMixin, BelongsToOrgMixin, db.Model): schedule = Column(db.String(10), nullable=True) schedule_failures = Column(db.Integer, default=0) schedule_until = Column(db.DateTime(True), nullable=True) + schedule_resultset_size = Column(db.Integer, nullable=True) visualizations = db.relationship("Visualization", cascade="all, delete-orphan") options = Column(MutableDict.as_mutable(PseudoJSON), default={}) search_vector = Column(TSVectorType('id', 'name', 'description', 'query', @@ -1035,6 +1041,37 @@ def search(cls, term, group_ids, user_id=None, include_drafts=False, limit=None) # sort the result using the weight as defined in the search vector column return all_queries.search(term, sort=True).limit(limit) + @classmethod + def delete_stale_resultsets(cls): + delete_count = 0 + texts = [c[0] for c in db.session.query(Query.query_text) + .filter(Query.schedule_resultset_size != None).distinct()] + for text in texts: + queries = (Query.query.filter(Query.query_text == text, + Query.schedule_resultset_size != None) + .order_by(Query.schedule_resultset_size.desc())) + # Multiple queries with the same text may request multiple result sets + # be kept. We start with the one that keeps the most, and delete both + # the unneeded bridge rows and result sets. + first_query = queries.first() + if first_query is not None and first_query.schedule_resultset_size: + resultsets = QueryResultSet.query.filter(QueryResultSet.query_rel == first_query).order_by(QueryResultSet.result_id) + resultset_count = resultsets.count() + if resultset_count > first_query.schedule_resultset_size: + n_to_delete = resultset_count - first_query.schedule_resultset_size + r_ids = [r.result_id for r in resultsets][:n_to_delete] + QueryResultSet.query.filter(QueryResultSet.result_id.in_(r_ids)).delete(synchronize_session=False) + delete_count += QueryResult.query.filter(QueryResult.id.in_(r_ids)).delete(synchronize_session=False) + # By this point there are no stale result sets left. + # Delete unneeded bridge rows for the remaining queries. + for q in queries[1:]: + resultsets = db.session.query(QueryResultSet.result_id).filter(QueryResultSet.query_rel == q).order_by(QueryResultSet.result_id) + n_to_delete = resultsets.count() - q.schedule_resultset_size + if n_to_delete > 0: + stale_r = QueryResultSet.query.filter(QueryResultSet.result_id.in_(resultsets.limit(n_to_delete).subquery())) + stale_r.delete(synchronize_session=False) + return delete_count + @classmethod def search_by_user(cls, term, user, limit=None): return cls.by_user(user).search(term, sort=True).limit(limit) @@ -1116,6 +1153,16 @@ def __repr__(self): return '' % (self.id, self.name or 'untitled') +class QueryResultSet(db.Model): + query_id = Column(db.Integer, db.ForeignKey("queries.id"), + primary_key=True) + query_rel = db.relationship(Query) + result_id = Column(db.Integer, db.ForeignKey("query_results.id"), + primary_key=True) + result = db.relationship(QueryResult) + __tablename__ = 'query_resultsets' + + @vectorizer(db.Integer) def integer_vectorizer(column): return db.func.cast(column, db.Text) diff --git a/redash/serializers.py b/redash/serializers.py index 21ca3d1238..44f43ae454 100644 --- a/redash/serializers.py +++ b/redash/serializers.py @@ -100,6 +100,7 @@ def serialize_query(query, with_stats=False, with_visualizations=False, with_use 'query_hash': query.query_hash, 'schedule': query.schedule, 'schedule_until': query.schedule_until, + 'schedule_resultset_size': query.schedule_resultset_size, 'api_key': query.api_key, 'is_archived': query.is_archived, 'is_draft': query.is_draft, diff --git a/redash/tasks/queries.py b/redash/tasks/queries.py index e6c3a61006..77c01c15aa 100644 --- a/redash/tasks/queries.py +++ b/redash/tasks/queries.py @@ -355,6 +355,7 @@ def cleanup_query_results(): deleted_count = models.QueryResult.query.filter( models.QueryResult.id.in_(unused_query_results.subquery()) ).delete(synchronize_session=False) + deleted_count += models.Query.delete_stale_resultsets() models.db.session.commit() logger.info("Deleted %d unused query results.", deleted_count) diff --git a/tests/factories.py b/tests/factories.py index 15f85e58cb..00423fdfa7 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -109,7 +109,9 @@ def __call__(self): query_hash=gen_query_hash('SELECT 1'), data_source=data_source_factory.create, org_id=1) - +query_resultset_factory = ModelFactory(redash.models.QueryResultSet, + query_rel=query_factory.create, + result=query_result_factory.create) visualization_factory = ModelFactory(redash.models.Visualization, type='CHART', query_rel=query_factory.create, @@ -295,6 +297,9 @@ def create_query_result(self, **kwargs): return query_result_factory.create(**args) + def create_query_resultset(self, **kwargs): + return query_resultset_factory.create(**kwargs) + def create_visualization(self, **kwargs): args = { 'query_rel': self.create_query() diff --git a/tests/handlers/test_queries.py b/tests/handlers/test_queries.py index 93dfae83d0..135d29c69a 100644 --- a/tests/handlers/test_queries.py +++ b/tests/handlers/test_queries.py @@ -1,3 +1,5 @@ +import json + from tests import BaseTestCase from redash import models from redash.models import db @@ -285,3 +287,81 @@ def test_get(self): rv2 = self.make_request('get', '/api/changes/' + str(ch2.id)) self.assertEqual(rv2.status_code, 200) self.assertEqual(rv2.json['change']['name']['current'], 'version B') + + +class AggregateResultsTests(BaseTestCase): + def test_aggregate(self): + qtxt = "SELECT x FROM mytable;" + q = self.factory.create_query(query_text=qtxt, schedule_resultset_size=3) + qr0 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'eve', 'color': 'grue'}, + {'name': 'mallory', 'color': 'bleen'}]})) + qr1 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'bob', 'color': 'green'}, + {'name': 'fred', 'color': 'blue'}]})) + qr2 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'alice', 'color': 'red'}, + {'name': 'eddie', 'color': 'orange'}]})) + qr3 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'dave', 'color': 'yellow'}, + {'name': 'carol', 'color': 'taupe'}]})) + for qr in (qr0, qr1, qr2, qr3): + self.factory.create_query_resultset(query_rel=q, result=qr) + rv = self.make_request('get', '/api/queries/{}/resultset'.format(q.id)) + self.assertEqual(rv.status_code, 200) + self.assertEqual(rv.json['query_result']['data'], + {'columns': ['name', 'color'], + 'rows': [ + {'name': 'bob', 'color': 'green'}, + {'name': 'fred', 'color': 'blue'}, + {'name': 'alice', 'color': 'red'}, + {'name': 'eddie', 'color': 'orange'}, + {'name': 'dave', 'color': 'yellow'}, + {'name': 'carol', 'color': 'taupe'} + ]}) + + def test_underfilled_aggregate(self): + qtxt = "SELECT x FROM mytable;" + q = self.factory.create_query(query_text=qtxt, + schedule_resultset_size=3) + qr1 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'bob', 'color': 'green'}, + {'name': 'fred', 'color': 'blue'}]})) + qr2 = self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'alice', 'color': 'red'}, + {'name': 'eddie', 'color': 'orange'}]})) + for qr in (qr1, qr2): + self.factory.create_query_resultset(query_rel=q, result=qr) + rv = self.make_request('get', '/api/queries/{}/resultset'.format(q.id)) + self.assertEqual(rv.status_code, 200) + self.assertEqual(rv.json['query_result']['data'], + {'columns': ['name', 'color'], + 'rows': [ + {'name': 'bob', 'color': 'green'}, + {'name': 'fred', 'color': 'blue'}, + {'name': 'alice', 'color': 'red'}, + {'name': 'eddie', 'color': 'orange'} + ]}) + + def test_no_aggregate(self): + qtxt = "SELECT x FROM mytable;" + q = self.factory.create_query(query_text=qtxt) + self.factory.create_query_result( + query_text=qtxt, + data=json.dumps({'columns': ['name', 'color'], + 'rows': [{'name': 'eve', 'color': 'grue'}, + {'name': 'mallory', 'color': 'bleen'}]})) + rv = self.make_request('get', '/api/queries/{}/resultset'.format(q.id)) + self.assertEqual(rv.status_code, 404) diff --git a/tests/test_models.py b/tests/test_models.py index 4d15eae932..8767b67fcf 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -335,22 +335,74 @@ def test_get_latest_returns_the_last_cached_result_for_negative_ttl(self): class TestUnusedQueryResults(BaseTestCase): def test_returns_only_unused_query_results(self): two_weeks_ago = utcnow() - datetime.timedelta(days=14) - qr = self.factory.create_query_result() - query = self.factory.create_query(latest_query_data=qr) + qt = "SELECT 1" + qr = self.factory.create_query_result(query_text=qt, retrieved_at=two_weeks_ago) + query = self.factory.create_query(query_text=qt, latest_query_data=qr) + unused_qr = self.factory.create_query_result(query_text=qt, retrieved_at=two_weeks_ago) db.session.flush() - unused_qr = self.factory.create_query_result(retrieved_at=two_weeks_ago) self.assertIn((unused_qr.id,), models.QueryResult.unused()) self.assertNotIn((qr.id,), list(models.QueryResult.unused())) def test_returns_only_over_a_week_old_results(self): two_weeks_ago = utcnow() - datetime.timedelta(days=14) - unused_qr = self.factory.create_query_result(retrieved_at=two_weeks_ago) + qt = "SELECT 1" + unused_qr = self.factory.create_query_result(query_text=qt, retrieved_at=two_weeks_ago) db.session.flush() - new_unused_qr = self.factory.create_query_result() - + new_unused_qr = self.factory.create_query_result(query_text=qt) self.assertIn((unused_qr.id,), models.QueryResult.unused()) self.assertNotIn((new_unused_qr.id,), models.QueryResult.unused()) + def test_doesnt_return_live_incremental_results(self): + two_weeks_ago = utcnow() - datetime.timedelta(days=14) + qt = "SELECT 1" + qrs = [self.factory.create_query_result(query_text=qt, retrieved_at=two_weeks_ago) + for _ in range(5)] + q = self.factory.create_query(query_text=qt, latest_query_data=qrs[0], + schedule_resultset_size=3) + for qr in qrs: + self.factory.create_query_resultset(query_rel=q, result=qr) + db.session.flush() + self.assertEqual([], list(models.QueryResult.unused())) + + def test_deletes_stale_resultsets(self): + qt = "SELECT 17" + query = self.factory.create_query(query_text=qt, + schedule_resultset_size=5) + for _ in range(10): + r = self.factory.create_query_result(query_text=qt) + self.factory.create_query_resultset(query_rel=query, result=r) + qt2 = "SELECT 100" + query2 = self.factory.create_query(query_text=qt2, schedule_resultset_size=5) + for _ in range(10): + r = self.factory.create_query_result(query_text=qt2) + self.factory.create_query_resultset(query_rel=query2, result=r) + db.session.flush() + self.assertEqual(models.QueryResultSet.query.count(), 20) + self.assertEqual(models.Query.delete_stale_resultsets(), 10) + self.assertEqual(models.QueryResultSet.query.count(), 10) + + def test_deletes_stale_resultsets_with_dupe_queries(self): + qt = "SELECT 17" + query = self.factory.create_query(query_text=qt, + schedule_resultset_size=5) + for _ in range(10): + r = self.factory.create_query_result(query_text=qt) + self.factory.create_query_resultset(query_rel=query, result=r) + query2 = self.factory.create_query(query_text=qt, + schedule_resultset_size=3) + for _ in range(10): + self.factory.create_query_result(query_text=qt) + self.factory.create_query_resultset(query_rel=query2) + qt2 = "SELECT 100" + query3 = self.factory.create_query(query_text=qt2, schedule_resultset_size=5) + for _ in range(10): + r = self.factory.create_query_result(query_text=qt2) + self.factory.create_query_resultset(query_rel=query3, result=r) + db.session.flush() + self.assertEqual(models.QueryResultSet.query.count(), 30) + self.assertEqual(models.Query.delete_stale_resultsets(), 10) + self.assertEqual(models.QueryResultSet.query.count(), 13) + class TestQueryAll(BaseTestCase): def test_returns_only_queries_in_given_groups(self):