Pārlūkot izejas kodu

[catalog] Move the analyze and describe fetch function to Axios

This also takes care of an issue with describe on a database and improves overall error messages for catalog related API calls.
Johan Ahlen 5 gadi atpakaļ
vecāks
revīzija
24c9f31ebb

+ 0 - 119
desktop/core/src/desktop/js/api/apiHelper.js

@@ -1285,57 +1285,6 @@ class ApiHelper {
     return simplePost(url, data, options);
   }
 
-  /**
-   * Fetches the analysis for the given source and path
-   *
-   * @param {Object} options
-   * @param {boolean} [options.silenceErrors]
-   *
-   * @param {ContextCompute} options.compute
-   * @param {string} options.sourceType
-   * @param {string[]} options.path
-   *
-   * @return {CancellableJqPromise}
-   */
-  fetchAnalysis(options) {
-    const deferred = $.Deferred();
-
-    let url = '/notebook/api/describe/' + options.path[0];
-
-    if (options.path.length > 1) {
-      url += '/' + options.path[1] + '/';
-    }
-
-    if (options.path.length > 2) {
-      url += 'stats/' + options.path.slice(2).join('/');
-    }
-
-    const data = {
-      format: 'json',
-      cluster: JSON.stringify(options.compute),
-      source_type: options.sourceType
-    };
-
-    const request = simplePost(url, data, {
-      silenceErrors: options.silenceErrors,
-      successCallback: response => {
-        if (options.path.length === 1) {
-          if (response.data) {
-            response.data.hueTimestamp = Date.now();
-            deferred.resolve(response.data);
-          } else {
-            deferred.reject();
-          }
-        } else {
-          deferred.resolve(response);
-        }
-      },
-      errorCallback: deferred.reject
-    });
-
-    return new CancellableJqPromise(deferred, request);
-  }
-
   /**
    * Fetches the partitions for the given path
    *
@@ -1392,74 +1341,6 @@ class ApiHelper {
     return new CancellableJqPromise(deferred, request);
   }
 
-  /**
-   * Refreshes the analysis for the given source and path
-   *
-   * @param {Object} options
-   * @param {boolean} [options.silenceErrors]
-   *
-   * @param {string} options.sourceType
-   * @param {ContextCompute} options.compute
-   * @param {string[]} options.path
-   *
-   * @return {CancellableJqPromise}
-   */
-  refreshAnalysis(options) {
-    if (options.path.length === 1) {
-      return this.fetchAnalysis(options);
-    }
-    const deferred = $.Deferred();
-
-    const promises = [];
-
-    const pollForAnalysis = (url, delay) => {
-      window.setTimeout(() => {
-        promises.push(
-          simplePost(url, undefined, {
-            silenceErrors: options.silenceErrors,
-            successCallback: data => {
-              promises.pop();
-              if (data.isSuccess) {
-                promises.push(
-                  this.fetchAnalysis(options).done(deferred.resolve).fail(deferred.reject)
-                );
-              } else if (data.isFailure) {
-                deferred.reject(data);
-              } else {
-                pollForAnalysis(url, 1000);
-              }
-            },
-            errorCallback: deferred.reject
-          })
-        );
-      }, delay);
-    };
-
-    const url =
-      '/' +
-      (options.sourceType === 'hive' ? 'beeswax' : options.sourceType) +
-      '/api/analyze/' +
-      options.path.join('/') +
-      '/';
-
-    promises.push(
-      simplePost(url, undefined, {
-        silenceErrors: options.silenceErrors,
-        successCallback: data => {
-          promises.pop();
-          if (data.status === 0 && data.watch_url) {
-            pollForAnalysis(data.watch_url, 500);
-          } else {
-            deferred.reject();
-          }
-        },
-        errorCallback: deferred.reject
-      })
-    );
-
-    return new CancellableJqPromise(deferred, undefined, promises);
-  }
-
   /**
    * Checks the status for the given snippet ID
    * Note: similar to notebook and search check_status.

+ 15 - 19
desktop/core/src/desktop/js/catalog/DataCatalogEntry.ts

@@ -15,7 +15,7 @@
 // limitations under the License.
 
 import { Cancellable, CancellablePromise } from 'api/cancellablePromise';
-import { fetchNavigatorMetadata, fetchSourceMetadata } from 'catalog/api';
+import { fetchDescribe, fetchNavigatorMetadata, fetchSourceMetadata } from 'catalog/api';
 import MultiTableEntry, { TopAggs, TopFilters, TopJoins } from 'catalog/MultiTableEntry';
 import { getOptimizer } from './optimizer/optimizer';
 import * as ko from 'knockout';
@@ -289,31 +289,27 @@ const reloadNavigatorMeta = (
  */
 const reloadAnalysis = (
   entry: DataCatalogEntry,
-  apiOptions?: { silenceErrors?: boolean; refreshAnalysis?: boolean }
+  options?: { silenceErrors?: boolean; refreshAnalysis?: boolean }
 ): CancellablePromise<Analysis> => {
-  entry.analysisPromise = new CancellablePromise<Analysis>((resolve, reject, onCancel) => {
-    const apiFunction =
-      apiOptions && apiOptions.refreshAnalysis
-        ? apiHelper.refreshAnalysis.bind(apiHelper)
-        : apiHelper.fetchAnalysis.bind(apiHelper);
-
-    const fetchPromise = fetchAndSave(
-      (<(options: FetchOptions) => CancellableJqPromise<Analysis>>(<unknown>apiFunction)).bind(
-        apiHelper
-      ),
-      val => {
-        entry.analysis = val;
-      },
+  entry.analysisPromise = new CancellablePromise<Analysis>(async (resolve, reject, onCancel) => {
+    const fetchPromise = fetchDescribe({
       entry,
-      apiOptions
-    );
+      ...options
+    });
 
     onCancel(() => {
       fetchPromise.cancel();
-      entry.analysis = undefined;
+      entry.analysisPromise = undefined;
     });
 
-    fetchPromise.then(resolve).fail(reject);
+    try {
+      entry.analysis = await fetchPromise;
+      resolve(entry.analysis);
+    } catch (err) {
+      reject(err || 'Fetch failed');
+      return;
+    }
+    entry.saveLater();
   });
   return entry.analysisPromise;
 };

+ 172 - 54
desktop/core/src/desktop/js/catalog/api.ts

@@ -16,94 +16,212 @@
 
 import { CancellablePromise } from 'api/cancellablePromise';
 import { DefaultApiResponse, extractErrorMessage, post, successResponseIsError } from 'api/utils';
-import DataCatalogEntry, { NavigatorMeta, SourceMeta } from 'catalog/DataCatalogEntry';
+import DataCatalogEntry, { Analysis, NavigatorMeta, SourceMeta } from 'catalog/DataCatalogEntry';
+import { sleep } from 'utils/hueUtils';
 
-const AUTOCOMPLETE_URL_PREFIX = '/notebook/api/autocomplete/';
-const FIND_ENTITY_URL = '/metadata/api/catalog/find_entity';
+interface AnalyzeResponse {
+  status: number;
+  isSuccess: boolean;
+  isFailure: boolean;
+}
 
-export const fetchSourceMetadata = (options: {
+interface SharedFetchOptions {
   entry: DataCatalogEntry;
   silenceErrors?: boolean;
-}): CancellablePromise<SourceMeta> =>
-  post<SourceMeta>(
-    `${AUTOCOMPLETE_URL_PREFIX}${options.entry.path.join('/')}${
-      options.entry.path.length ? '/' : ''
-    }`,
-    {
-      notebook: {},
-      snippet: JSON.stringify({
-        type: options.entry.getConnector().id,
-        source: 'data'
-      }),
-      cluster: (options.entry.compute && JSON.stringify(options.entry.compute)) || '""'
-    },
-    {
-      ...options,
-      handleResponse: response => {
-        const message = <string>response.error || response.message || '';
-        const adjustedResponse = response;
-        adjustedResponse.notFound =
-          response.status === 0 &&
-          response.code === 500 &&
-          (message.indexOf('Error 10001') !== -1 || message.indexOf('AnalysisException') !== -1);
+}
 
-        adjustedResponse.hueTimestamp = Date.now();
+const AUTOCOMPLETE_URL_PREFIX = '/notebook/api/autocomplete/';
+const DESCRIBE_URL = '/notebook/api/describe/';
+const FIND_ENTITY_URL = '/metadata/api/catalog/find_entity';
+
+const performAnalyze = ({
+  entry,
+  silenceErrors
+}: SharedFetchOptions): CancellablePromise<AnalyzeResponse> => {
+  if (entry.isDatabase()) {
+    return CancellablePromise.resolve();
+  }
+  let cancelled = false;
+
+  const pollForAnalysis = async (url: string, delay: number): Promise<AnalyzeResponse> => {
+    const analyzeResponse = await post<AnalyzeResponse>(url, undefined, { silenceErrors });
+    if (cancelled) {
+      throw new Error('Cancelled');
+    }
+    if (!analyzeResponse.isFailure && !analyzeResponse.isSuccess) {
+      await sleep(delay);
+      return pollForAnalysis(url, 1000);
+    } else {
+      return analyzeResponse;
+    }
+  };
+
+  return new CancellablePromise<AnalyzeResponse>(async (resolve, reject, onCancel) => {
+    onCancel(() => {
+      cancelled = true;
+    });
+    try {
+      const analyzeResponse = await post<DefaultApiResponse & { watch_url?: string }>(
+        `/${
+          entry.getConnector().id === 'hive' ? 'beeswax' : entry.getConnector().id
+        }/api/analyze/${entry.path.join('/')}/`,
+        undefined,
+        { silenceErrors }
+      );
+      if (
+        !cancelled &&
+        analyzeResponse &&
+        analyzeResponse.status === 0 &&
+        analyzeResponse.watch_url
+      ) {
+        resolve(await pollForAnalysis(analyzeResponse.watch_url, 500));
+      } else {
+        reject('Analyze failed');
+      }
+    } catch (err) {
+      reject(err || 'Analyze failed');
+    }
+  });
+};
+
+export const fetchDescribe = ({
+  entry,
+  silenceErrors,
+  refreshAnalysis
+}: SharedFetchOptions & {
+  refreshAnalysis?: boolean;
+}): CancellablePromise<Analysis> =>
+  new CancellablePromise<Analysis>(async (resolve, reject, onCancel) => {
+    if (entry.isSource()) {
+      reject('Describe is not possible on the source');
+      return;
+    }
+
+    if (refreshAnalysis) {
+      const analyzePromise = performAnalyze({ entry, silenceErrors });
+      onCancel(analyzePromise.cancel.bind(analyzePromise));
+      try {
+        await analyzePromise;
+      } catch (err) {}
+    }
 
-        const valid = adjustedResponse.notFound || !successResponseIsError(response);
+    const [database, table, ...fields] = entry.path;
+    let url = `${DESCRIBE_URL}${database}`;
+    if (table && fields.length) {
+      url += `/${table}/stats/${fields.join('/')}`;
+    } else if (table) {
+      url += `/${table}/`;
+    }
 
-        if (!valid) {
-          return { valid, reason: extractErrorMessage(response) };
+    const describePromise = post<Analysis>(
+      url,
+      {
+        format: 'json',
+        cluster: JSON.stringify(entry.compute),
+        source_type: entry.getConnector().id
+      },
+      {
+        silenceErrors,
+        handleResponse: (response: Analysis & DefaultApiResponse) => {
+          if (successResponseIsError(response)) {
+            return { valid: false, reason: extractErrorMessage(response) };
+          }
+          const adjustedResponse = response;
+          adjustedResponse.hueTimestamp = Date.now();
+          return { valid: true, adjustedResponse };
         }
-        return { valid, adjustedResponse };
       }
+    );
+
+    try {
+      resolve(await describePromise);
+    } catch (err) {
+      reject(err || 'Describe failed');
     }
-  );
+  });
 
-export const fetchNavigatorMetadata = (options: {
-  entry: DataCatalogEntry;
-  silenceErrors?: boolean;
-}): CancellablePromise<NavigatorMeta> => {
+export const fetchNavigatorMetadata = ({
+  entry,
+  silenceErrors
+}: SharedFetchOptions): CancellablePromise<NavigatorMeta> => {
   const params = new URLSearchParams();
 
-  if (options.entry.path.length === 1) {
-    params.append('type', 'database');
-  } else if (options.entry.path.length === 2) {
-    params.append('type', options.entry.isView() ? 'view' : 'table');
-    params.append('database', options.entry.path[0]);
-  } else if (options.entry.path.length === 3) {
+  const [database, tableOrView, field] = entry.path;
+  if (database && tableOrView && field) {
     params.append('type', 'field');
-    params.append('database', options.entry.path[1]);
-    params.append('table', options.entry.path[1]);
+    params.append('database', database);
+    params.append('table', tableOrView);
+  } else if (database && tableOrView) {
+    params.append('type', entry.isView() ? 'view' : 'table');
+    params.append('database', database);
+  } else if (database) {
+    params.append('type', 'database');
   } else {
-    return CancellablePromise.reject();
+    return CancellablePromise.reject('Navigator metadata is not possible on the source');
   }
-  params.append('name', options.entry.path[options.entry.path.length - 1]);
+  params.append('name', entry.name);
 
   return post<NavigatorMeta>(
     `${FIND_ENTITY_URL}?${params}`,
     {
       notebook: {},
       snippet: JSON.stringify({
-        type: options.entry.getConnector().id,
+        type: entry.getConnector().id,
         source: 'data'
       }),
-      cluster: (options.entry.compute && JSON.stringify(options.entry.compute)) || '""'
+      cluster: (entry.compute && JSON.stringify(entry.compute)) || '""'
     },
     {
-      ...options,
+      silenceErrors,
       handleResponse: (
         response: (NavigatorMeta | { entity: NavigatorMeta }) & DefaultApiResponse
       ) => {
+        if (successResponseIsError(response)) {
+          return {
+            valid: false,
+            reason: `Navigator meta failed: ${extractErrorMessage(response)}`
+          };
+        }
         const adjustedResponse = (<{ entity: NavigatorMeta }>response).entity || response;
         adjustedResponse.hueTimestamp = Date.now();
 
-        const valid = !successResponseIsError(response);
+        return { valid: true, adjustedResponse };
+      }
+    }
+  );
+};
 
-        if (!valid) {
-          return { valid, reason: extractErrorMessage(response) };
+export const fetchSourceMetadata = ({
+  entry,
+  silenceErrors
+}: SharedFetchOptions): CancellablePromise<SourceMeta> =>
+  post<SourceMeta>(
+    `${AUTOCOMPLETE_URL_PREFIX}${entry.path.join('/')}${entry.path.length ? '/' : ''}`,
+    {
+      notebook: {},
+      snippet: JSON.stringify({
+        type: entry.getConnector().id,
+        source: 'data'
+      }),
+      cluster: (entry.compute && JSON.stringify(entry.compute)) || '""'
+    },
+    {
+      silenceErrors,
+      handleResponse: response => {
+        const message = <string>response.error || response.message || '';
+        const adjustedResponse = response || {};
+        adjustedResponse.notFound =
+          !!response &&
+          response.status === 0 &&
+          response.code === 500 &&
+          (message.indexOf('Error 10001') !== -1 || message.indexOf('AnalysisException') !== -1);
+
+        adjustedResponse.hueTimestamp = Date.now();
+
+        if (!adjustedResponse.notFound && successResponseIsError(response)) {
+          return { valid: false, reason: `Source meta failed: ${extractErrorMessage(response)}` };
         }
-        return { valid, adjustedResponse };
+        return { valid: true, adjustedResponse };
       }
     }
   );
-};