Line data    Source code

       1              : import { DataFrame, generateTableName } from '../data_frame.js';
       2              : import { TableFilteringTask, FilterTable } from '../computation_types.js';
       3              : import { LoggableException, LoggerLike, stringifyError } from '../../platform/logger/logger.js';
       4              : import { SQLFrame } from '../sql/sqlframe_builder.js';
       5              : import { ComputeQueryExecution, executeDirectly } from '../computation_tasks.js';
       6              : 
       7            1 : const LOG_CTX = "compute";
       8              : 
       9            1 : export async function filterTable(task: TableFilteringTask, logger: LoggerLike, executeQuery: ComputeQueryExecution = executeDirectly): Promise<FilterTable | null> {
      10            1 :     if (task.filters.length == 0) {
      11            1 :         return null;
      12              :     }
      13              : 
      14            1 :     try {
      15            1 :         let frame = SQLFrame.from(task.inputDataFrame.tableName);
      16            1 :         for (const f of task.filters) {
      17            1 :             frame = frame.filter(f.fieldName, f.op, f.value);
      18              :         }
      19            1 :         frame = frame.project([task.rowNumberColumnName]);
      20            1 :         const sql = frame.toSQL();
      21              : 
      22            1 :         logger.info("Filtering table", {
      23            1 :             "version": task.tableVersion.toString(),
      24            1 :         }, LOG_CTX);
      25            1 :         const filterStart = performance.now();
      26            1 :         const tableName = generateTableName("__filter");
      27            1 :         const transformed = await executeQuery(sql, clientContext => DataFrame.fromSQL(task.inputDataFrame.database, sql, tableName, undefined, clientContext));
      28            1 :         const filterEnd = performance.now();
      29              :         // The rows of the filter are shown in the order of the input
      30            1 :         const filterResultTable = await transformed.readTable({ orderBy: [{ field: task.rowNumberColumnName }] });
      31              : 
      32            1 :         logger.info("Filtered table", {
      33            1 :             "version": task.tableVersion.toString(),
      34            1 :             "duration": Math.floor(filterEnd - filterStart).toString(),
      35            1 :             "inputRows": task.inputDataTable.numRows.toString(),
      36            1 :             "outputRows": filterResultTable.numRows.toString(),
      37            1 :             "sql": sql,
      38            1 :         }, LOG_CTX);
      39              : 
      40              :         const out: FilterTable = {
      41            1 :             inputRowNumberColumnName: task.rowNumberColumnName,
      42            1 :             dataTable: filterResultTable,
      43            1 :             dataFrame: transformed,
      44              :             // Version is already the target - state was incremented when task was scheduled
      45            1 :             version: task.tableVersion,
      46              :         };
      47            1 :         return out;
      48              : 
      49            1 :     } catch (error: any) {
      50            0 :         if (error instanceof LoggableException) {
      51            0 :             throw error;
      52              :         } else {
      53            0 :             throw new LoggableException(`Failed to filter table`, { "error": stringifyError(error) }, LOG_CTX);
      54              :         }
      55            0 :     }
      56            1 : }