feat(SIP-39): Async query support for charts (#11499)

* Generate JWT in Flask app

* Refactor chart data API query logic, add JWT validation and async worker

* Add redis stream implementation, refactoring

* Add chart data cache endpoint, refactor QueryContext caching

* Typing, linting, refactoring

* pytest fixes and openapi schema update

* Enforce caching be configured for async query init

* Async query processing for explore_json endpoint

* Add /api/v1/async_event endpoint

* Async frontend for dashboards [WIP]

* Chart async error message support, refactoring

* Abstract asyncEvent middleware

* Async chart loading for Explore

* Pylint fixes

* asyncEvent middleware -> TypeScript, JS linting

* Chart data API: enforce forced_cache, add tests

* Add tests for explore_json endpoints

* Add test for chart data cache enpoint (no login)

* Consolidate set_and_log_cache and add STORE_CACHE_KEYS_IN_METADATA_DB flag

* Add tests for tasks/async_queries and address PR comments

* Bypass non-JSON result formats for async queries

* Add tests for redux middleware

* Remove debug statement

Co-authored-by: Ville Brofeldt <[email protected]>

* Skip force_cached if no queryObj

* SunburstViz: don't modify self.form_data

* Fix failing annotation test

* Resolve merge/lint issues

* Reduce polling delay

* Fix new getClientErrorObject reference

* Fix flakey unit tests

* /api/v1/async_event: increment redis stream ID, add tests

* PR feedback: refactoring, configuration

* Fixup: remove debugging

* Fix typescript errors due to redux upgrade

* Update UPDATING.md

* Fix failing py tests

* asyncEvent_spec.js -> asyncEvent_spec.ts

* Refactor flakey Python 3.7 mock assertions

* Fix another shared state issue in Py tests

* Use 'sub' claim in JWT for user_id

* Refactor async middleware config

* Fixup: restore FeatureFlag boolean type

Co-authored-by: Ville Brofeldt <[email protected]>

CONTRIBUTING.md

@@ -623,6 +623,11 @@ cd superset-frontend
 npm run test
 ```
 
+To run a single test file:
+```bash
+npm run test -- path/to/file.js
+```
+
 ### Integration Testing
 
 We use [Cypress](https://www.cypress.io/) for integration tests. Tests can be run by `tox -e cypress`. To open Cypress and explore tests first setup and run test server:

UPDATING.md

@@ -23,7 +23,7 @@ This file documents any backwards-incompatible changes in Superset and
 assists people when migrating to a new version.
 
 ## Next
-
+- [11499](https://github.com/apache/incubator-superset/pull/11499): Breaking change: `STORE_CACHE_KEYS_IN_METADATA_DB` config flag added (default=`False`) to write `CacheKey` records to the metadata DB. `CacheKey` recording was enabled by default previously.
 - [11920](https://github.com/apache/incubator-superset/pull/11920): Undos the DB migration from [11714](https://github.com/apache/incubator-superset/pull/11714) to prevent adding new columns to the logs table. Deploying a sha between these two PRs may result in locking your DB.
 - [11704](https://github.com/apache/incubator-superset/pull/11704) Breaking change: Jinja templating for SQL queries has been updated, removing default modules such as `datetime` and `random` and enforcing static template values. To restore or extend functionality, use `JINJA_CONTEXT_ADDONS` and `CUSTOM_TEMPLATE_PROCESSORS` in `superset_config.py`.
 - [11714](https://github.com/apache/incubator-superset/pull/11714): Logs

setup.cfg

@@ -30,7 +30,7 @@ combine_as_imports = true
 include_trailing_comma = true
 line_length = 88
 known_first_party = superset
-known_third_party =alembic,apispec,backoff,bleach,cachelib,celery,click,colorama,contextlib2,croniter,cryptography,dateutil,flask,flask_appbuilder,flask_babel,flask_caching,flask_compress,flask_login,flask_migrate,flask_sqlalchemy,flask_talisman,flask_testing,flask_wtf,freezegun,geohash,geopy,humanize,isodate,jinja2,markdown,markupsafe,marshmallow,msgpack,numpy,pandas,parameterized,parsedatetime,pathlib2,pgsanity,polyline,prison,pyarrow,pyhive,pytest,pytz,retry,selenium,setuptools,simplejson,slack,sqlalchemy,sqlalchemy_utils,sqlparse,werkzeug,wtforms,wtforms_json,yaml
+known_third_party =alembic,apispec,backoff,bleach,cachelib,celery,click,colorama,contextlib2,croniter,cryptography,dateutil,flask,flask_appbuilder,flask_babel,flask_caching,flask_compress,flask_login,flask_migrate,flask_sqlalchemy,flask_talisman,flask_testing,flask_wtf,freezegun,geohash,geopy,humanize,isodate,jinja2,jwt,markdown,markupsafe,marshmallow,msgpack,numpy,pandas,parameterized,parsedatetime,pathlib2,pgsanity,polyline,prison,pyarrow,pyhive,pytest,pytz,redis,retry,selenium,setuptools,simplejson,slack,sqlalchemy,sqlalchemy_utils,sqlparse,werkzeug,wtforms,wtforms_json,yaml
 multi_line_output = 3
 order_by_type = false
 

superset-frontend/spec/javascripts/middleware/asyncEvent_spec.ts

@@ -0,0 +1,265 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import fetchMock from 'fetch-mock';
+import sinon from 'sinon';
+import * as featureFlags from 'src/featureFlags';
+import initAsyncEvents from 'src/middleware/asyncEvent';
+
+jest.useFakeTimers();
+
+describe('asyncEvent middleware', () => {
+  const next = sinon.spy();
+  const state = {
+    charts: {
+      123: {
+        id: 123,
+        status: 'loading',
+        asyncJobId: 'foo123',
+      },
+      345: {
+        id: 345,
+        status: 'loading',
+        asyncJobId: 'foo345',
+      },
+    },
+  };
+  const events = [
+    {
+      status: 'done',
+      result_url: '/api/v1/chart/data/cache-key-1',
+      job_id: 'foo123',
+      channel_id: '999',
+      errors: [],
+    },
+    {
+      status: 'done',
+      result_url: '/api/v1/chart/data/cache-key-2',
+      job_id: 'foo345',
+      channel_id: '999',
+      errors: [],
+    },
+  ];
+  const mockStore = {
+    getState: () => state,
+    dispatch: sinon.stub(),
+  };
+  const action = {
+    type: 'GENERIC_ACTION',
+  };
+  const EVENTS_ENDPOINT = 'glob:*/api/v1/async_event/*';
+  const CACHED_DATA_ENDPOINT = 'glob:*/api/v1/chart/data/*';
+  const config = {
+    GLOBAL_ASYNC_QUERIES_TRANSPORT: 'polling',
+    GLOBAL_ASYNC_QUERIES_POLLING_DELAY: 500,
+  };
+  let featureEnabledStub: any;
+
+  function setup() {
+    const getPendingComponents = sinon.stub();
+    const successAction = sinon.spy();
+    const errorAction = sinon.spy();
+    const testCallback = sinon.stub();
+    const testCallbackPromise = sinon.stub();
+    testCallbackPromise.returns(
+      new Promise(resolve => {
+        testCallback.callsFake(resolve);
+      }),
+    );
+
+    return {
+      getPendingComponents,
+      successAction,
+      errorAction,
+      testCallback,
+      testCallbackPromise,
+    };
+  }
+
+  beforeEach(() => {
+    fetchMock.get(EVENTS_ENDPOINT, {
+      status: 200,
+      body: { result: [] },
+    });
+    fetchMock.get(CACHED_DATA_ENDPOINT, {
+      status: 200,
+      body: { result: { some: 'data' } },
+    });
+    featureEnabledStub = sinon.stub(featureFlags, 'isFeatureEnabled');
+    featureEnabledStub.withArgs('GLOBAL_ASYNC_QUERIES').returns(true);
+  });
+  afterEach(() => {
+    fetchMock.reset();
+    next.resetHistory();
+    featureEnabledStub.restore();
+  });
+  afterAll(fetchMock.reset);
+
+  it('should initialize and call next', () => {
+    const { getPendingComponents, successAction, errorAction } = setup();
+    getPendingComponents.returns([]);
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+    });
+    asyncEventMiddleware(mockStore)(next)(action);
+    expect(next.callCount).toBe(1);
+  });
+
+  it('should fetch events when there are pending components', () => {
+    const {
+      getPendingComponents,
+      successAction,
+      errorAction,
+      testCallback,
+      testCallbackPromise,
+    } = setup();
+    getPendingComponents.returns(Object.values(state.charts));
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+      processEventsCallback: testCallback,
+    });
+
+    asyncEventMiddleware(mockStore)(next)(action);
+
+    return testCallbackPromise().then(() => {
+      expect(fetchMock.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+    });
+  });
+
+  it('should fetch cached when there are successful events', () => {
+    const {
+      getPendingComponents,
+      successAction,
+      errorAction,
+      testCallback,
+      testCallbackPromise,
+    } = setup();
+    fetchMock.reset();
+    fetchMock.get(EVENTS_ENDPOINT, {
+      status: 200,
+      body: { result: events },
+    });
+    fetchMock.get(CACHED_DATA_ENDPOINT, {
+      status: 200,
+      body: { result: { some: 'data' } },
+    });
+    getPendingComponents.returns(Object.values(state.charts));
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+      processEventsCallback: testCallback,
+    });
+
+    asyncEventMiddleware(mockStore)(next)(action);
+
+    return testCallbackPromise().then(() => {
+      expect(fetchMock.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+      expect(fetchMock.calls(CACHED_DATA_ENDPOINT)).toHaveLength(2);
+      expect(successAction.callCount).toBe(2);
+    });
+  });
+
+  it('should call errorAction for cache fetch error responses', () => {
+    const {
+      getPendingComponents,
+      successAction,
+      errorAction,
+      testCallback,
+      testCallbackPromise,
+    } = setup();
+    fetchMock.reset();
+    fetchMock.get(EVENTS_ENDPOINT, {
+      status: 200,
+      body: { result: events },
+    });
+    fetchMock.get(CACHED_DATA_ENDPOINT, {
+      status: 400,
+      body: { errors: ['error'] },
+    });
+    getPendingComponents.returns(Object.values(state.charts));
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+      processEventsCallback: testCallback,
+    });
+
+    asyncEventMiddleware(mockStore)(next)(action);
+
+    return testCallbackPromise().then(() => {
+      expect(fetchMock.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+      expect(fetchMock.calls(CACHED_DATA_ENDPOINT)).toHaveLength(2);
+      expect(errorAction.callCount).toBe(2);
+    });
+  });
+
+  it('should handle event fetching error responses', () => {
+    const {
+      getPendingComponents,
+      successAction,
+      errorAction,
+      testCallback,
+      testCallbackPromise,
+    } = setup();
+    fetchMock.reset();
+    fetchMock.get(EVENTS_ENDPOINT, {
+      status: 400,
+      body: { message: 'error' },
+    });
+    getPendingComponents.returns(Object.values(state.charts));
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+      processEventsCallback: testCallback,
+    });
+
+    asyncEventMiddleware(mockStore)(next)(action);
+
+    return testCallbackPromise().then(() => {
+      expect(fetchMock.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+    });
+  });
+
+  it('should not fetch events when async queries are disabled', () => {
+    featureEnabledStub.restore();
+    featureEnabledStub = sinon.stub(featureFlags, 'isFeatureEnabled');
+    featureEnabledStub.withArgs('GLOBAL_ASYNC_QUERIES').returns(false);
+    const { getPendingComponents, successAction, errorAction } = setup();
+    getPendingComponents.returns(Object.values(state.charts));
+    const asyncEventMiddleware = initAsyncEvents({
+      config,
+      getPendingComponents,
+      successAction,
+      errorAction,
+    });
+
+    asyncEventMiddleware(mockStore)(next)(action);
+    expect(getPendingComponents.called).toBe(false);
+  });
+});

superset-frontend/spec/javascripts/utils/getClientErrorObject_spec.ts

@@ -17,7 +17,7 @@
  * under the License.
  */
 import { ErrorTypeEnum } from 'src/components/ErrorMessage/types';
-import getClientErrorObject from 'src/utils/getClientErrorObject';
+import { getClientErrorObject } from 'src/utils/getClientErrorObject';
 
 describe('getClientErrorObject()', () => {
   it('Returns a Promise', () => {

superset-frontend/src/SqlLab/actions/sqlLab.js

@@ -30,7 +30,7 @@ import {
   addSuccessToast as addSuccessToastAction,
   addWarningToast as addWarningToastAction,
 } from '../../messageToasts/actions/index';
-import getClientErrorObject from '../../utils/getClientErrorObject';
+import { getClientErrorObject } from '../../utils/getClientErrorObject';
 import COMMON_ERR_MESSAGES from '../../utils/errorMessages';
 
 export const RESET_STATE = 'RESET_STATE';

superset-frontend/src/SqlLab/components/ShareSqlLabQuery.jsx

@@ -25,7 +25,7 @@ import { isFeatureEnabled, FeatureFlag } from 'src/featureFlags';
 import Button from 'src/components/Button';
 import CopyToClipboard from '../../components/CopyToClipboard';
 import { storeQuery } from '../../utils/common';
-import getClientErrorObject from '../../utils/getClientErrorObject';
+import { getClientErrorObject } from '../../utils/getClientErrorObject';
 import withToasts from '../../messageToasts/enhancers/withToasts';
 
 const propTypes = {

superset-frontend/src/chart/chartAction.js

@@ -38,7 +38,7 @@ import {
 import { addDangerToast } from '../messageToasts/actions';
 import { logEvent } from '../logger/actions';
 import { Logger, LOG_ACTIONS_LOAD_CHART } from '../logger/LogUtils';
-import getClientErrorObject from '../utils/getClientErrorObject';
+import { getClientErrorObject } from '../utils/getClientErrorObject';
 import { allowCrossDomain as domainShardingEnabled } from '../utils/hostNamesConfig';
 
 export const CHART_UPDATE_STARTED = 'CHART_UPDATE_STARTED';
@@ -66,6 +66,11 @@ export function chartUpdateFailed(queryResponse, key) {
   return { type: CHART_UPDATE_FAILED, queryResponse, key };
 }
 
+export const CHART_UPDATE_QUEUED = 'CHART_UPDATE_QUEUED';
+export function chartUpdateQueued(asyncJobMeta, key) {
+  return { type: CHART_UPDATE_QUEUED, asyncJobMeta, key };
+}
+
 export const CHART_RENDERING_FAILED = 'CHART_RENDERING_FAILED';
 export function chartRenderingFailed(error, key, stackTrace) {
   return { type: CHART_RENDERING_FAILED, error, key, stackTrace };
@@ -356,6 +361,12 @@ export function exploreJSON(
 
     const chartDataRequestCaught = chartDataRequest
       .then(response => {
+        if (isFeatureEnabled(FeatureFlag.GLOBAL_ASYNC_QUERIES)) {
+          // deal with getChartDataRequest transforming the response data
+          const result = 'result' in response ? response.result[0] : response;
+          return dispatch(chartUpdateQueued(result, key));
+        }
+
         // new API returns an object with an array of restults
         // problem: response holds a list of results, when before we were just getting one result.
         // How to make the entire app compatible with multiple results?

superset-frontend/src/chart/chartReducer.js

@@ -71,6 +71,14 @@ export default function chartReducer(charts = {}, action) {
         chartUpdateEndTime: now(),
       };
     },
+    [actions.CHART_UPDATE_QUEUED](state) {
+      return {
+        ...state,
+        asyncJobId: action.asyncJobMeta.job_id,
+        chartStatus: 'loading',
+        chartUpdateEndTime: now(),
+      };
+    },
     [actions.CHART_RENDERING_SUCCEEDED](state) {
       return { ...state, chartStatus: 'rendered', chartUpdateEndTime: now() };
     },

superset-frontend/src/components/AsyncSelect.jsx

@@ -21,7 +21,7 @@ import PropTypes from 'prop-types';
 // TODO: refactor this with `import { AsyncSelect } from src/components/Select`
 import { Select } from 'src/components/Select';
 import { t, SupersetClient } from '@superset-ui/core';
-import getClientErrorObject from '../utils/getClientErrorObject';
+import { getClientErrorObject } from '../utils/getClientErrorObject';
 
 const propTypes = {
   dataEndpoint: PropTypes.string.isRequired,

superset-frontend/src/dashboard/actions/dashboardState.js

@@ -29,7 +29,7 @@ import {
   updateDirectPathToFilter,
 } from './dashboardFilters';
 import { applyDefaultFormData } from '../../explore/store';
-import getClientErrorObject from '../../utils/getClientErrorObject';
+import { getClientErrorObject } from '../../utils/getClientErrorObject';
 import { SAVE_TYPE_OVERWRITE } from '../util/constants';
 import {
   addSuccessToast,

superset-frontend/src/dashboard/actions/datasources.js

@@ -17,7 +17,7 @@
  * under the License.
  */
 import { SupersetClient } from '@superset-ui/core';
-import getClientErrorObject from '../../utils/getClientErrorObject';
+import { getClientErrorObject } from '../../utils/getClientErrorObject';
 
 export const SET_DATASOURCE = 'SET_DATASOURCE';
 export function setDatasource(datasource, key) {