Line data    Source code

       1              : import { DataFrame, generateTableName } from '../data_frame.js';
       2              : import { TableOrderingTask, OrderingTable } 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 sortTable(task: TableOrderingTask, logger: LoggerLike, executeQuery: ComputeQueryExecution = executeDirectly): Promise<OrderingTable> {
      10            1 :     if (task.orderingConstraints.length == 1) {
      11            1 :         logger.info("Sorting table by field", {
      12            1 :             "field": task.orderingConstraints[0].field
      13            1 :         }, LOG_CTX);
      14              :     } else {
      15            0 :         logger.info("Sorting table by multiple fields", {}, LOG_CTX);
      16              :     }
      17              : 
      18            1 :     try {
      19            1 :         let frame = SQLFrame.from(task.inputDataFrame.tableName);
      20            1 :         if (task.filterTable != null) {
      21            0 :             frame = frame.semiJoinFilter(
      22            0 :                 task.rowNumberColumnName,
      23            0 :                 task.filterTable.dataFrame.tableName,
      24            0 :                 task.filterTable.dataTable.schema.fields[0].name,
      25            0 :             );
      26              :         }
      27            1 :         if (task.dataSearchTable != null) {
      28            0 :             frame = frame.semiJoinFilter(
      29            0 :                 task.rowNumberColumnName,
      30            0 :                 task.dataSearchTable.dataFrame.tableName,
      31            0 :                 task.dataSearchTable.dataTable.schema.fields[0].name,
      32            0 :             );
      33              :         }
      34              :         // The table keeps the fields of the ordering, the read sorts the rows. Equal rows keep the order of the input.
      35            1 :         const fields = new Set([task.rowNumberColumnName, ...task.orderingConstraints.map(constraint => constraint.field)]);
      36            1 :         const sql = frame
      37            1 :             .project([...fields])
      38            1 :             .toSQL();
      39              : 
      40            1 :         const sortStart = performance.now();
      41            1 :         const tableName = generateTableName("__ordered");
      42            1 :         const transformed = await executeQuery(sql, clientContext => DataFrame.fromSQL(task.inputDataFrame.database, sql, tableName, undefined, clientContext));
      43            1 :         const orderedTable = await transformed.readTable({
      44            1 :             orderBy: [...task.orderingConstraints, { field: task.rowNumberColumnName }],
      45            1 :             columns: [task.rowNumberColumnName],
      46            1 :         });
      47            1 :         const sortEnd = performance.now();
      48            1 :         logger.info("Sorted table", {
      49            1 :             "version": task.tableVersion.toString(),
      50            1 :             "duration": Math.floor(sortEnd - sortStart).toString(),
      51            1 :             "inputRows": task.inputDataTable.numRows.toString(),
      52            1 :             "outputRows": orderedTable.numRows.toString(),
      53            1 :         }, LOG_CTX);
      54              :         const out: OrderingTable = {
      55            1 :             inputRowNumberColumnName: task.rowNumberColumnName,
      56            1 :             orderingConstraints: task.orderingConstraints,
      57            1 :             dataTable: orderedTable,
      58            1 :             dataFrame: transformed,
      59              :             // Ordering depends on filtered data, so version stays same
      60            1 :             version: task.tableVersion,
      61            1 :             dataSearchRequestId: task.dataSearchTable?.requestId ?? null,
      62              :         };
      63            1 :         return out;
      64              : 
      65            1 :     } catch (error: any) {
      66            0 :         if (error instanceof LoggableException) {
      67            0 :             throw error;
      68              :         } else {
      69            0 :             throw new LoggableException(`Sorting table failed`, { "error": stringifyError(error) }, LOG_CTX);
      70              :         }
      71            0 :     }
      72            1 : }