Răsfoiți Sursa

[catalog] Move the samples fetch function to Axios

Johan Ahlen 5 ani în urmă
părinte
comite
322fd1abf8

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

@@ -29,7 +29,6 @@ import apiQueueManager from 'api/apiQueueManager';
 import CancellableJqPromise from 'api/cancellableJqPromise';
 import hueDebug from 'utils/hueDebug';
 import huePubSub from 'utils/huePubSub';
-import hueUtils from 'utils/hueUtils';
 import { getFromLocalStorage, setInLocalStorage } from 'utils/storageUtils';
 
 export const LINK_SHARING_PERMS = {
@@ -38,25 +37,6 @@ export const LINK_SHARING_PERMS = {
   OFF: 'off'
 };
 
-/**
- * Wrapper around the response from the Query API
- *
- * @param {string} sourceType
- * @param {Object} response
- *
- * @constructor
- */
-class QueryResult {
-  constructor(sourceType, compute, response) {
-    this.id = hueUtils.UUID();
-    this.type = response.result && response.result.type ? response.result.type : sourceType;
-    this.compute = compute;
-    this.status = response.status || 'running';
-    this.result = response.result || {};
-    this.result.type = 'table';
-  }
-}
-
 class ApiHelper {
   constructor() {
     this.queueManager = apiQueueManager;
@@ -1341,62 +1321,6 @@ class ApiHelper {
     return new CancellableJqPromise(deferred, request);
   }
 
-  /**
-   * Checks the status for the given snippet ID
-   * Note: similar to notebook and search check_status.
-   *
-   * @param {Object} options
-   * @param {Object} options.notebookJson
-   * @param {Object} options.snippetJson
-   * @param {boolean} [options.silenceErrors]
-   *
-   * @return {CancellableJqPromise}
-   */
-  whenAvailable(options) {
-    const deferred = $.Deferred();
-    const cancellablePromises = [];
-
-    let waitTimeout = -1;
-
-    deferred.fail(() => {
-      window.clearTimeout(waitTimeout);
-    });
-
-    const waitForAvailable = () => {
-      const request = simplePost(
-        '/notebook/api/check_status',
-        {
-          notebook: options.notebookJson,
-          snippet: options.snippetJson,
-          cluster: ko.mapping.toJSON(options.compute ? options.compute : '""')
-        },
-        {
-          silenceErrors: options.silenceErrors
-        }
-      )
-        .done(response => {
-          if (response && response.query_status && response.query_status.status) {
-            const status = response.query_status.status;
-            if (status === 'available') {
-              deferred.resolve(response.query_status);
-            } else if (status === 'running' || status === 'starting' || status === 'waiting') {
-              waitTimeout = window.setTimeout(() => {
-                waitForAvailable();
-              }, 500);
-            } else {
-              deferred.reject(response.query_status);
-            }
-          }
-        })
-        .fail(deferred.reject);
-
-      cancellablePromises.push(new CancellableJqPromise(request, request));
-    };
-
-    waitForAvailable();
-    return new CancellableJqPromise(deferred, undefined, cancellablePromises);
-  }
-
   clearNotebookHistory(options) {
     const data = {
       notebook: options.notebookJson,
@@ -1542,133 +1466,6 @@ class ApiHelper {
     });
   }
 
-  /**
-   * Fetches samples for the given source and path
-   *
-   * @param {Object} options
-   * @param {boolean} [options.silenceErrors]
-   *
-   * @param {string} options.sourceType
-   * @param {ContextCompute} options.compute
-   * @param {number} [options.sampleCount] - Default 100
-   * @param {string[]} options.path
-   * @param {string} [options.operation] - Default 'default'
-   *
-   * @return {CancellableJqPromise}
-   */
-  fetchSample(options) {
-    const deferred = $.Deferred();
-
-    const cancellablePromises = [];
-
-    let notebookJson = null;
-    let snippetJson = null;
-
-    const cancelQuery = () => {
-      if (notebookJson) {
-        simplePost(
-          '/notebook/api/cancel_statement',
-          {
-            notebook: notebookJson,
-            snippet: snippetJson,
-            cluster: ko.mapping.toJSON(options.compute ? options.compute : '""')
-          },
-          { silenceErrors: options.silenceErrors }
-        );
-      }
-    };
-
-    simplePost(
-      URLS.SAMPLE_API_PREFIX + options.path.join('/') + (options.path.length ? '/' : ''),
-      {
-        notebook: {},
-        snippet: JSON.stringify({
-          type: options.sourceType,
-          compute: options.compute
-        }),
-        async: true,
-        operation: '"' + (options.operation || 'default') + '"',
-        cluster: ko.mapping.toJSON(options.compute ? options.compute : '""')
-      },
-      {
-        silenceErrors: options.silenceErrors
-      }
-    )
-      .done(sampleResponse => {
-        const queryResult = new QueryResult(options.sourceType, options.compute, sampleResponse);
-
-        notebookJson = JSON.stringify({ type: options.sourceType });
-        snippetJson = JSON.stringify(queryResult);
-
-        if (sampleResponse && sampleResponse.rows) {
-          // Sync results
-          const data = { data: sampleResponse.rows, meta: sampleResponse.full_headers };
-          data.hueTimestamp = Date.now();
-          deferred.resolve(data);
-        } else {
-          cancellablePromises.push(
-            this.whenAvailable({
-              notebookJson: notebookJson,
-              snippetJson: snippetJson,
-              compute: options.compute,
-              silenceErrors: options.silenceErrors
-            })
-              .done(resultStatus => {
-                if (resultStatus) {
-                  $.extend(true, queryResult.result, {
-                    handle: resultStatus
-                  });
-                }
-                const resultRequest = simplePost(
-                  '/notebook/api/fetch_result_data',
-                  {
-                    notebook: notebookJson,
-                    snippet: JSON.stringify(queryResult),
-                    rows: options.sampleCount || 100,
-                    startOver: 'false'
-                  },
-                  {
-                    silenceErrors: options.silenceErrors
-                  }
-                )
-                  .done(sampleResponse => {
-                    const data = (sampleResponse && sampleResponse.result) || {
-                      data: [],
-                      meta: []
-                    };
-                    data.hueTimestamp = Date.now();
-                    deferred.resolve(data);
-
-                    if (
-                      window.CLOSE_SESSIONS[options.sourceType] &&
-                      queryResult.result.handle &&
-                      queryResult.result.handle.session_id
-                    ) {
-                      this.closeSession({
-                        session: {
-                          type: options.sourceType,
-                          id: queryResult.result.handle.session_id
-                        }
-                      });
-                    }
-                  })
-                  .fail(deferred.reject);
-
-                cancellablePromises.push(resultRequest, resultRequest);
-              })
-              .fail(deferred.reject)
-          );
-        }
-      })
-      .fail(deferred.reject);
-
-    cancellablePromises.push({
-      cancel: cancelQuery
-    });
-
-    return new CancellableJqPromise(deferred, undefined, cancellablePromises);
-  }
-
   /**
    * Updates Navigator properties and custom metadata for the given entity
    *

+ 60 - 84
desktop/core/src/desktop/js/catalog/DataCatalogEntry.ts

@@ -19,6 +19,7 @@ import {
   fetchDescribe,
   fetchNavigatorMetadata,
   fetchPartitions,
+  fetchSample,
   fetchSourceMetadata
 } from 'catalog/api';
 import MultiTableEntry, { TopAggs, TopFilters, TopJoins } from 'catalog/MultiTableEntry';
@@ -223,57 +224,11 @@ export interface OptimizerMeta extends TimestampedData {
   hueTimestamp?: number;
 }
 
-const setAndSave = async <T extends keyof DataCatalogEntry>(
-  entry: DataCatalogEntry,
-  attribute: T,
-  promise: Promise<DataCatalogEntry[T]>,
-  resolve: (val: DataCatalogEntry[T]) => void,
-  reject: (reason: unknown) => void
-): Promise<void> => {
-  try {
-    entry[attribute] = await promise;
-    resolve(entry[attribute]);
-  } catch (err) {
-    reject(err || 'Fetch failed');
-    return;
-  }
-  entry.saveLater();
-};
-
 const cachedOnly = (options?: CatalogGetOptions): boolean => !!(options && options.cachedOnly);
 
 const shouldReload = (options?: CatalogGetOptions & { refreshAnalysis?: boolean }): boolean =>
   !!(!DataCatalog.cacheEnabled() || (options && (options.refreshCache || options.refreshAnalysis)));
 
-/**
- * Helper function to reload the sample for the given entry
- */
-const reloadSample = (
-  entry: DataCatalogEntry,
-  apiOptions?: { silenceErrors?: boolean }
-): CancellablePromise<Sample> => {
-  entry.samplePromise = new CancellablePromise<Sample>((resolve, reject, onCancel) => {
-    const fetchPromise = fetchAndSave(
-      (<(options: FetchOptions) => CancellableJqPromise<Sample>>(
-        (<unknown>apiHelper.fetchSample)
-      )).bind(apiHelper),
-      val => {
-        entry.sample = val;
-      },
-      entry,
-      apiOptions
-    );
-
-    onCancel(() => {
-      fetchPromise.cancel();
-      entry.samplePromise = undefined;
-    });
-
-    fetchPromise.then(resolve).fail(reject);
-  });
-  return entry.samplePromise;
-};
-
 /**
  * Helper function to reload the nav opt metadata for the given entry
  */
@@ -470,7 +425,14 @@ export default class DataCatalogEntry {
         this.analysisPromise = undefined;
       });
 
-      await setAndSave(this, 'analysis', fetchPromise, resolve, reject);
+      try {
+        this.analysis = await fetchPromise;
+        resolve(this.analysis);
+      } catch (err) {
+        reject(err || 'Fetch failed');
+        return;
+      }
+      this.saveLater();
     });
     return applyCancellable(this.analysisPromise, options);
   }
@@ -482,13 +444,14 @@ export default class DataCatalogEntry {
           onCancel(() => {
             this.navigatorMetaPromise = undefined;
           });
-          await setAndSave(
-            this,
-            'navigatorMeta',
-            fetchNavigatorMetadata({ ...options, entry: this }),
-            resolve,
-            reject
-          );
+          try {
+            this.navigatorMeta = await fetchNavigatorMetadata({ ...options, entry: this });
+            resolve(this.navigatorMeta);
+          } catch (err) {
+            reject(err || 'Fetch failed');
+            return;
+          }
+          this.saveLater();
           if (this.commentObservable) {
             this.commentObservable(this.getResolvedComment());
           }
@@ -507,18 +470,39 @@ export default class DataCatalogEntry {
           this.partitionsPromise = undefined;
         });
 
-        await setAndSave(
-          this,
-          'partitions',
-          fetchPartitions({ ...options, entry: this }),
-          resolve,
-          reject
-        );
+        try {
+          this.partitions = await fetchPartitions({ ...options, entry: this });
+          resolve(this.partitions);
+        } catch (err) {
+          reject(err || 'Fetch failed');
+          return;
+        }
+        this.saveLater();
       }
     );
     return applyCancellable(this.partitionsPromise, options);
   }
 
+  private reloadSample(
+    options?: ReloadOptions & { operation?: string }
+  ): CancellablePromise<Sample> {
+    this.samplePromise = new CancellablePromise<Sample>(async (resolve, reject, onCancel) => {
+      onCancel(() => {
+        this.samplePromise = undefined;
+      });
+
+      try {
+        this.sample = await fetchSample({ ...options, entry: this });
+        resolve(this.sample);
+      } catch (err) {
+        reject(err || 'Fetch failed');
+        return;
+      }
+      this.saveLater();
+    });
+    return applyCancellable(this.samplePromise, options);
+  }
+
   private reloadSourceMeta(options?: ReloadOptions): CancellablePromise<SourceMeta> {
     this.sourceMetaPromise = new CancellablePromise<SourceMeta>(
       async (resolve, reject, onCancel) => {
@@ -532,16 +516,17 @@ export default class DataCatalogEntry {
           } catch (err) {}
         }
 
-        await setAndSave(
-          this,
-          'sourceMeta',
-          fetchSourceMetadata({
+        try {
+          this.sourceMeta = await fetchSourceMetadata({
             ...options,
             entry: this
-          }),
-          resolve,
-          reject
-        );
+          });
+          resolve(this.sourceMeta);
+        } catch (err) {
+          reject(err || 'Fetch failed');
+          return;
+        }
+        this.saveLater();
       }
     );
     return applyCancellable(this.sourceMetaPromise, options);
@@ -1603,19 +1588,10 @@ export default class DataCatalogEntry {
     // This prevents caching of any non-standard sample queries, i.e. DISTINCT etc.
     if (options && options.operation && options.operation !== 'default') {
       const operation = options.operation;
-      const samplePromise = new CancellablePromise<Sample>((resolve, reject, onCancel) => {
-        const fetchPromise = apiHelper.fetchSample({
-          sourceType: this.getConnector().id,
-          compute: this.compute,
-          path: this.path,
-          silenceErrors: !!(options && options.silenceErrors),
-          operation
-        });
-        onCancel(() => {
-          fetchPromise.cancel();
-        });
-
-        fetchPromise.done(resolve).fail(reject);
+      const samplePromise = fetchSample({
+        entry: this,
+        operation,
+        silenceErrors: options.silenceErrors
       });
       return applyCancellable(samplePromise, options);
     }
@@ -1671,7 +1647,7 @@ export default class DataCatalogEntry {
         if (cachedOnly(options)) {
           reject();
         } else {
-          const reloadPromise = applyCancellable(reloadSample(this, options), options);
+          const reloadPromise = this.reloadSample(options);
           cancellablePromises.push(reloadPromise);
           try {
             resolve(await reloadPromise);
@@ -1688,7 +1664,7 @@ export default class DataCatalogEntry {
       return CancellablePromise.reject();
     }
     if (!this.samplePromise || shouldReload(options)) {
-      return applyCancellable(reloadSample(this, options), options);
+      return this.reloadSample(options);
     }
     return applyCancellable(this.samplePromise, options);
   }

+ 245 - 5
desktop/core/src/desktop/js/catalog/api.ts

@@ -14,15 +14,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import { CancellablePromise } from 'api/cancellablePromise';
+import { Cancellable, CancellablePromise } from 'api/cancellablePromise';
 import { DefaultApiResponse, extractErrorMessage, post, successResponseIsError } from 'api/utils';
+import { closeSession, ExecutionHandle } from 'apps/notebook2/execution/api';
 import DataCatalogEntry, {
   Analysis,
+  FieldSample,
   NavigatorMeta,
   Partitions,
+  Sample,
+  SampleMeta,
   SourceMeta
 } from 'catalog/DataCatalogEntry';
-import { sleep } from 'utils/hueUtils';
+import { hueWindow } from 'types/types';
+import { sleep, UUID } from 'utils/hueUtils';
 
 interface AnalyzeResponse {
   status: number;
@@ -35,10 +40,23 @@ interface SharedFetchOptions {
   silenceErrors?: boolean;
 }
 
+interface DescribeFetchOptions extends SharedFetchOptions {
+  refreshAnalysis?: boolean;
+}
+
+interface SampleFetchOptions extends SharedFetchOptions {
+  operation?: string;
+  sampleCount?: number;
+}
+
 const AUTOCOMPLETE_URL_PREFIX = '/notebook/api/autocomplete/';
+const CANCEL_STATEMENT_URL = '/notebook/api/cancel_statement';
+const CHECK_STATUS_URL = '/notebook/api/check_status';
 const DESCRIBE_URL = '/notebook/api/describe/';
+const FETCH_RESULT_DATA_URL = '/notebook/api/fetch_result_data';
 const FIND_ENTITY_URL = '/metadata/api/catalog/find_entity';
 const METASTORE_TABLE_URL_PREFIX = '/metastore/table/';
+const SAMPLE_URL_PREFIX = '/notebook/api/sample/';
 
 const getEntryUrlPath = (entry: DataCatalogEntry) =>
   entry.path.join('/') + (entry.path.length ? '/' : '');
@@ -97,9 +115,7 @@ export const fetchDescribe = ({
   entry,
   silenceErrors,
   refreshAnalysis
-}: SharedFetchOptions & {
-  refreshAnalysis?: boolean;
-}): CancellablePromise<Analysis> =>
+}: DescribeFetchOptions): CancellablePromise<Analysis> =>
   new CancellablePromise<Analysis>(async (resolve, reject, onCancel) => {
     if (entry.isSource()) {
       reject('Describe is not possible on the source');
@@ -239,6 +255,230 @@ export const fetchPartitions = ({
     }
   );
 
+interface SampleResult {
+  type?: string;
+  handle?: ExecutionHandle;
+  data?: FieldSample[][];
+  meta?: SampleMeta[];
+}
+
+interface SampleResponse {
+  status?: string;
+  result?: SampleResult;
+  rows?: FieldSample[][];
+  full_headers?: SampleMeta[];
+}
+
+/**
+ * Checks the status for the given snippet ID
+ * Note: similar to notebook and search check_status.
+ *
+ * @param {Object} options
+ * @param {Object} options.notebookJson
+ * @param {Object} options.snippetJson
+ * @param {boolean} [options.silenceErrors]
+ *
+ * @return {CancellableJqPromise}
+ */
+const whenAvailable = (options: {
+  entry: DataCatalogEntry;
+  notebookJson: string;
+  snippetJson: string;
+  silenceErrors?: boolean;
+}) =>
+  new CancellablePromise<{ status?: string }>(async (resolve, reject, onCancel) => {
+    let promiseToCancel: Cancellable | undefined;
+    let cancelled = false;
+    onCancel(() => {
+      cancelled = true;
+      if (promiseToCancel) {
+        promiseToCancel.cancel();
+      }
+    });
+
+    const checkStatusPromise = post<{ query_status?: { status?: string } }>(
+      CHECK_STATUS_URL,
+      {
+        notebook: options.notebookJson,
+        snippet: options.snippetJson,
+        cluster: (options.entry.compute && JSON.stringify(options.entry.compute)) || '""'
+      },
+      { silenceErrors: options.silenceErrors }
+    );
+    try {
+      promiseToCancel = checkStatusPromise;
+      const response = await checkStatusPromise;
+
+      if (response && response.query_status && response.query_status.status) {
+        const status = response.query_status.status;
+        if (status === 'available') {
+          resolve(response.query_status);
+        } else if (status === 'running' || status === 'starting' || status === 'waiting') {
+          await sleep(500);
+          try {
+            if (!cancelled) {
+              const whenPromise = whenAvailable(options);
+              promiseToCancel = whenPromise;
+              resolve(await whenPromise);
+              return;
+            }
+          } catch (err) {}
+        }
+        reject(response.query_status);
+      } else {
+        reject('Cancelled');
+      }
+    } catch (err) {
+      reject(err);
+    }
+  });
+
+export const fetchSample = ({
+  entry,
+  silenceErrors,
+  operation,
+  sampleCount
+}: SampleFetchOptions): CancellablePromise<Sample> =>
+  new CancellablePromise<Sample>(async (resolve, reject, onCancel) => {
+    const cancellablePromises: Cancellable[] = [];
+
+    let notebookJson: string | undefined = undefined;
+    let snippetJson: string | undefined = undefined;
+
+    const cancelQuery = async () => {
+      if (notebookJson) {
+        try {
+          await post(
+            CANCEL_STATEMENT_URL,
+            {
+              notebook: notebookJson,
+              snippet: snippetJson,
+              cluster: (entry.compute && JSON.stringify(entry.compute)) || '""'
+            },
+            { silenceErrors: true }
+          );
+        } catch (err) {}
+      }
+    };
+
+    onCancel(() => {
+      cancellablePromises.forEach(cancellable => cancellable.cancel());
+    });
+
+    cancellablePromises.push({
+      cancel: async () => {
+        try {
+          await cancelQuery();
+        } catch (err) {}
+      }
+    });
+
+    const samplePromise = post<SampleResponse>(
+      `${SAMPLE_URL_PREFIX}${getEntryUrlPath(entry)}`,
+      {
+        notebook: {},
+        snippet: JSON.stringify({
+          type: entry.getConnector().id,
+          compute: entry.compute
+        }),
+        async: true,
+        operation: `"${operation || 'default'}"`,
+        cluster: (entry.compute && JSON.stringify(entry.compute)) || '""'
+      },
+      { silenceErrors }
+    );
+
+    try {
+      cancellablePromises.push(samplePromise);
+      const sampleResponse = await samplePromise;
+      cancellablePromises.pop();
+
+      const queryResult = {
+        id: UUID(),
+        type: (sampleResponse.result && sampleResponse.result.type) || entry.getConnector().id,
+        compute: entry.compute,
+        status: 'running',
+        result: sampleResponse.result || {}
+      };
+      queryResult.result.type = 'table';
+
+      notebookJson = JSON.stringify({ type: entry.getConnector().id });
+      snippetJson = JSON.stringify(queryResult);
+
+      if (sampleResponse && sampleResponse.rows) {
+        // Sync results
+        resolve({
+          type: 'table',
+          hueTimestamp: Date.now(),
+          data: sampleResponse.rows,
+          meta: sampleResponse.full_headers || []
+        });
+      } else {
+        const statusPromise = whenAvailable({
+          notebookJson: notebookJson,
+          snippetJson: snippetJson,
+          entry,
+          silenceErrors
+        });
+
+        cancellablePromises.push(statusPromise);
+        const resultStatus = await statusPromise;
+        cancellablePromises.pop();
+
+        if (resultStatus.status !== 'available') {
+          reject();
+          return;
+        }
+
+        snippetJson = JSON.stringify(queryResult);
+        const resultPromise = post<SampleResponse>(
+          FETCH_RESULT_DATA_URL,
+          {
+            notebook: notebookJson,
+            snippet: snippetJson,
+            rows: sampleCount || 100,
+            startOver: 'false'
+          },
+          { silenceErrors }
+        );
+
+        const sampleResponse = await resultPromise;
+
+        const sample: Sample = {
+          hueTimestamp: Date.now(),
+          type: 'table',
+          data: (sampleResponse.result && sampleResponse.result.data) || [],
+          meta: (sampleResponse.result && sampleResponse.result.meta) || []
+        };
+
+        resolve(sample);
+        cancellablePromises.pop();
+
+        const closeSessions = (<hueWindow>window).CLOSE_SESSIONS;
+        if (
+          closeSessions &&
+          closeSessions[entry.getConnector().dialect || ''] &&
+          queryResult.result.handle &&
+          queryResult.result.handle.session_id
+        ) {
+          try {
+            await closeSession({
+              session: {
+                id: queryResult.result.handle.session_id,
+                session_id: queryResult.result.handle.session_guid || '',
+                type: entry.getConnector().id,
+                properties: []
+              },
+              silenceErrors
+            });
+          } catch (err) {}
+        }
+      }
+    } catch (err) {
+      reject();
+    }
+  });
+
 export const fetchSourceMetadata = ({
   entry,
   silenceErrors

+ 1 - 0
desktop/core/src/desktop/js/types/types.ts

@@ -32,6 +32,7 @@ declare global {
 
 export interface hueWindow {
   CACHEABLE_TTL?: { default?: number; optimizer?: number };
+  CLOSE_SESSIONS?: { [dialect: string]: boolean };
   HAS_CATALOG?: boolean;
   HAS_OPTIMIZER?: boolean;
   AUTOCOMPLETE_TIMEOUT?: number;