comments

querybook/server/datasources/datadoc.py

@@ -16,6 +16,7 @@
 from lib.celery.cron import validate_cron
 from lib.logger import get_logger
 from lib.notify.all_notifiers import DEFAULT_NOTIFIER
+from lib.notify.utils import get_user_preferred_notifier, notify_user
 from lib.scheduled_datadoc.validator import validate_datadoc_schedule_config
 from lib.scheduled_datadoc.legacy import convert_if_legacy_datadoc_schedule
 
@@ -32,7 +33,6 @@
     update_datadoc_schedule_owner,
 )
 from models.environment import Environment
-from lib.notify.utils import notify_user
 
 LOG = get_logger(__file__)
 
@@ -382,38 +382,29 @@ def run_data_doc(id):
 
 
 @register("/datadoc/<int:id>/run/", methods=["POST"])
-def adhoc_run_data_doc(id, originator=None):
-    with DBSession() as session:
-        assert_can_write(id, session=session)
-        verify_data_doc_permission(id, session=session)
-
-        notification_preference = user_logic.get_user_settings(
-            current_user.id, "notification_preference", session=session
-        )
-        notifier_name = (
-            notification_preference.value
-            if notification_preference is not None
-            else DEFAULT_NOTIFIER
-        )
-
-        result = celery.send_task(
-            "tasks.run_datadoc.run_datadoc",
-            args=[],
-            kwargs={
-                "doc_id": id,
-                "user_id": current_user.id,
-                "execution_type": QueryExecutionType.ADHOC.value,
-                "notifications": [
-                    {
-                        "config": {"to_user": [current_user.id]},
-                        "on": 0,
-                        "with": notifier_name,
-                    }
-                ],
-                "originator": originator,
-            },
-        )
-        return result
+def adhoc_run_data_doc(id):
+    assert_can_write(id)
+    verify_data_doc_permission(id)
+
+    notifier_name = get_user_preferred_notifier(current_user.id)
+
+    result = celery.send_task(
+        "tasks.run_datadoc.run_datadoc",
+        args=[],
+        kwargs={
+            "doc_id": id,
+            "user_id": current_user.id,
+            "execution_type": QueryExecutionType.ADHOC.value,
+            "notifications": [
+                {
+                    "config": {"to_user": [current_user.id]},
+                    "on": 0,
+                    "with": notifier_name,
+                }
+            ],
+        },
+    )
+    return result
 
 
 @register("/datadoc/<int:doc_id>/editor/", methods=["GET"])

querybook/server/lib/notify/utils.py

@@ -4,6 +4,18 @@
 from app.db import with_session
 
 
+@with_session
+def get_user_preferred_notifier(user_id, session=None):
+    notification_preference = user_logic.get_user_settings(
+        user_id, "notification_preference", session=session
+    )
+    return (
+        notification_preference.value
+        if notification_preference is not None
+        else DEFAULT_NOTIFIER
+    )
+
+
 def notify_recipients(recipients, template_name, template_params, notifier_name):
     notifier = get_notifier_class(notifier_name)
     markdown_message = render_message(template_name, template_params)
@@ -13,14 +25,7 @@ def notify_recipients(recipients, template_name, template_params, notifier_name)
 @with_session
 def notify_user(user, template_name, template_params, notifier_name=None, session=None):
     if notifier_name is None:
-        notification_preference = user_logic.get_user_settings(
-            user.id, "notification_preference", session=session
-        )
-        notifier_name = (
-            notification_preference.value
-            if notification_preference is not None
-            else DEFAULT_NOTIFIER
-        )
+        notifier_name = get_user_preferred_notifier(user.id, session=session)
         if notifier_name is None:
             return
     notifier = get_notifier_class(notifier_name)

querybook/server/tasks/run_datadoc.py

@@ -41,7 +41,6 @@ def run_datadoc_with_config(
     execution_type=QueryExecutionType.SCHEDULED.value,
     # Exporting related settings
     exports=[],
-    originator=None,
     *args,
     **kwargs,
 ):
@@ -92,7 +91,6 @@ def run_datadoc_with_config(
                     "uid": runner_id,
                 },
                 "data_doc_id": doc_id,
-                "originator": originator,
             }
             tasks_to_run.append(
                 _start_query_execution_task.si(
@@ -120,7 +118,6 @@ def _start_query_execution_task(
     cell_id,
     query_execution_params,
     data_doc_id,
-    originator,
 ):
     if previous_query_status != QueryExecutionStatus.DONE.value:
         raise Exception(GENERIC_QUERY_FAILURE_MSG)
@@ -135,20 +132,18 @@ def _start_query_execution_task(
             session=session,
         )
 
-        # adhoc datadoc run will pass over originator for realtime status update through websocket
-        if originator:
-            socketio.emit(
-                "data_doc_query_execution",
-                (
-                    originator,
-                    query_execution.to_dict(),
-                    cell_id,
-                    True,  # True here indicates the message is from data doc run (fromDataDocRun).
-                ),
-                namespace="/datadoc",
-                room=data_doc_id,
-                broadcast=True,
-            )
+        socketio.emit(
+            "data_doc_query_execution",
+            (
+                None,
+                query_execution.to_dict(),
+                cell_id,
+                True,  # True here indicates the message is from data doc run (fromDataDocRun).
+            ),
+            namespace="/datadoc",
+            room=data_doc_id,
+            broadcast=True,
+        )
         return query_execution.id
 
 
@@ -212,6 +207,7 @@ def on_datadoc_completion(
         error_msg = str(e)
         LOG.error(e, exc_info=True)
     finally:
+        # when record_id is None, it's trigerred by adhoc datadoc run, no need to update the record.
         if record_id:
             update_task_run_record(
                 id=record_id,

querybook/webapp/components/DataDocRightSidebar/DataDocRunAllButton.tsx

@@ -7,6 +7,7 @@ import { sendConfirm } from 'lib/querybookUI';
 import { makeLatestQueryExecutionsSelector } from 'redux/queryExecutions/selector';
 import { DataDocResource } from 'resource/dataDoc';
 import { IconButton } from 'ui/Button/IconButton';
+import { Message } from 'ui/Message/Message';
 
 interface IProps {
     docId: number;
@@ -20,30 +21,38 @@ export const DataDocRunAllButton: React.FunctionComponent<IProps> = ({
         makeLatestQueryExecutionsSelector,
         queryCells.map((c) => c.id) ?? []
     );
+    const hasQueryRunning = latestQueryExecutions.some((q) => q.status < 3);
 
-    const runAll = useCallback(
-        () =>
-            DataDocResource.run(docId).then(() => {
-                toast.success('DataDoc execution started!');
-            }),
-        [docId]
+    const ConfirmMessageDOM = useCallback(
+        () => (
+            <div>
+                <div>
+                    {`You will be executing ${queryCells.length} query cells sequentially. If any of them
+                fails, the sequence of execution will be stopped.`}
+                </div>
+                {hasQueryRunning && (
+                    <Message type="warning" className="mt8">
+                        There are some query cells still running. Do you want to
+                        run anyway?
+                    </Message>
+                )}
+            </div>
+        ),
+        [queryCells.length, hasQueryRunning]
     );
 
     const onRunAll = useCallback(() => {
-        if (latestQueryExecutions.some((q) => q.status < 3)) {
-            sendConfirm({
-                header: 'Run All Cells',
-                message:
-                    'At least one of the query cell is still running. Do you want to run anyway?',
-                onConfirm: () => {
-                    runAll();
-                },
-                confirmText: 'Run Anyway',
-            });
-        } else {
-            runAll();
-        }
-    }, [latestQueryExecutions, runAll]);
+        sendConfirm({
+            header: 'Run All Cells',
+            message: ConfirmMessageDOM(),
+            onConfirm: () => {
+                DataDocResource.run(docId).then(() => {
+                    toast.success('DataDoc execution started!');
+                });
+            },
+            confirmText: 'Run',
+        });
+    }, [ConfirmMessageDOM, docId]);
 
     return (
         <IconButton

querybook/webapp/redux/queryExecutions/selector.ts

@@ -92,15 +92,15 @@ export const queryExecutionViewersByUidSelector = createSelector(
             : {}
 );
 
-const latestQueryExecutionIdsSelector = (
+const latestQueryExecutionIdsPerCellSelector = (
     state: IStoreState,
     cellIds: number[]
 ) =>
     cellIds
         .map((cellId) => {
             const executions =
                 state.queryExecutions.dataCellIdQueryExecution[cellId] ?? [];
-            return [...executions].sort((a, b) => b - a)[0];
+            return Math.max(...executions);
         })
         .filter(Boolean);
 
@@ -109,7 +109,7 @@ const latestQueryExecutionIdsSelector = (
 // from the order of the input query cell ids.
 export const makeLatestQueryExecutionsSelector = () =>
     createSelector(
-        latestQueryExecutionIdsSelector,
+        latestQueryExecutionIdsPerCellSelector,
         queryExecutionByIdSelector,
         (queryExecutionIds, queryExecutionById) =>
             queryExecutionIds

querybook/webapp/resource/dataDoc.ts

@@ -101,10 +101,7 @@ export const DataDocResource = {
         }>(`/favorite_data_doc/${docId}/`),
     unfavorite: (docId: number) => ds.delete(`/favorite_data_doc/${docId}/`),
 
-    run: (docId: number) =>
-        ds.save<null>(`/datadoc/${docId}/run/`, {
-            originator: dataDocSocket.socketId,
-        }),
+    run: (docId: number) => ds.save<null>(`/datadoc/${docId}/run/`),
 
     getDAGExport: (docId: number) =>
         ds.fetch<IDataDocSavedDAGExport>(`/datadoc/${docId}/dag_export/`),