Aggregate query results (re getredash#35) (getredash#339)

Author: Allen Short

client/app/components/queries/schedule-dialog.html

@@ -19,4 +19,7 @@ <h4 class="modal-title">Refresh Schedule</h4>
       Stop scheduling at date/time (format yyyy-MM-ddTHH:mm:ss, like 2016-12-28T14:57:00):
       <schedule-until query="$ctrl.query" save-query="$ctrl.saveQuery"></schedule-until>
     </label>
+    <label>
+      Number of query results to keep <schedule-keep-results query="$ctrl.query" save-query="$ctrl.saveQuery"></schedule-keep-results>
+    </label>
 </div>

client/app/components/queries/schedule-dialog.js

@@ -114,11 +114,21 @@ function scheduleUntil() {
   };
 }
 
+function scheduleKeepResults() {
+  return {
+    restrict: 'E',
+    scope: {
+      query: '=',
+      saveQuery: '=',
+    },
+    template: '<input type="number" class="form-control" ng-model="query.schedule_resultset_size" ng-change="saveQuery()" ng-disabled="!query.schedule">',
+  };
+}
+
 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);
 }

client/app/pages/alerts-list/index.js

@@ -9,7 +9,6 @@ const stateClass = {
 
 class AlertsListCtrl {
   constructor(Alert) {
-
     this.showEmptyState = false;
     this.showList = false;
 

client/app/pages/queries/view.js

@@ -205,6 +205,7 @@ function QueryViewCtrl(
     } else {
       request = pick($scope.query, [
         'schedule',
+        'schedule_resultset_size',
         'query',
         'id',
         'description',

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(

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,

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 ###

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/<query_id>/refresh', endpoint='query_refresh')
 api.add_org_resource(QueryResource, '/api/queries/<query_id>', endpoint='query')
 api.add_org_resource(QueryForkResource, '/api/queries/<query_id>/fork', endpoint='query_fork')
+api.add_org_resource(QueryResultSetResource, '/api/queries/<query_id>/resultset', endpoint='query_aggregate_results')
 api.add_org_resource(QueryVersionListResource, '/api/queries/<query_id>/version', endpoint='query_versions')
 api.add_org_resource(ChangeResource, '/api/changes/<change_id>', endpoint='changes')
 

redash/handlers/queries.py

@@ -109,6 +109,7 @@ def post(self):
         :<json string description:
         :<json string schedule: Schedule interval, in seconds, for repeated execution of this query
         :<json string schedule_until: Time in ISO format to stop scheduling this query (may be null to run indefinitely)
+        :<json number schedule_resultset_size: Number of result sets to keep (null to keep only one)
         :<json object options: Query options
 
         .. _query-response-label:
@@ -146,6 +147,8 @@ def post(self):
         query_def['data_source'] = data_source
         query_def['org'] = self.current_org
         query_def['is_draft'] = True
+        if query_def.get('schedule_resultset_size') == 1:
+            query_def['schedule_resultset_size'] = None
         query = models.Query.create(**query_def)
         query.record_changes(changed_by=self.current_user)
         models.db.session.add(query)

redash/handlers/query_results.py

@@ -132,6 +132,33 @@ def post(self):
 ONE_YEAR = 60 * 60 * 24 * 365.25
 
 
+class QueryResultSetResource(BaseResource):
+    @require_permission('view_query')
+    def get(self, query_id=None, filetype='json'):
+        query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
+        if not query.schedule_resultset_size:
+            abort(404, message="query does not keep multiple results")
+
+        # Synthesize a result set from the last N results.
+        total = len(query.query_results)
+        offset = max(total - query.schedule_resultset_size, 0)
+        results = [qr.to_dict() for qr in query.query_results[offset:]]
+        if not results:
+            aggregate_result = {}
+        else:
+            # Start a synthetic data set with the data from the first result...
+            aggregate_result = results[0].copy()
+            aggregate_result['data'] = {'columns': results[0]['data']['columns'],
+                                        'rows': []}
+            # .. then add each subsequent result set into it.
+            for r in results:
+                aggregate_result['data']['rows'].extend(r['data']['rows'])
+
+        data = json.dumps({'query_result': aggregate_result}, cls=utils.JSONEncoder)
+        headers = {'Content-Type': "application/json"}
+        return make_response(data, 200, headers)
+
+
 class QueryResultResource(BaseResource):
     @staticmethod
     def add_cors_headers(headers):
@@ -194,7 +221,7 @@ def get(self, query_id=None, query_result_id=None, filetype='json'):
                     query_result = run_query_sync(query.data_source, parameter_values, query.query_text, max_age=max_age)
                 elif query.latest_query_data_id is not None:
                     query_result = get_object_or_404(models.QueryResult.get_by_id_and_org, query.latest_query_data_id, self.current_org)
-                
+
             if query is not None and query_result is not None and self.current_user.is_api_user():
                 if query.query_hash != query_result.query_hash:
                     abort(404, message='No cached result found for this query.')