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):