Line data Source code
1 : import * as arrow from 'apache-arrow';
2 :
3 : import { OrderByConstraint } from './sql/sqlframe_builder.js';
4 : import { ColumnAggregationVariant, TableAggregationTask, TableOrderingTask, TableAggregation, TaskProgress, ColumnGroup, SystemColumnComputationTask, FilterTable, ROWNUMBER_COLUMN, ORDINAL_COLUMN, STRING_COLUMN, LIST_COLUMN, SKIPPED_COLUMN, ColumnAggregationTask, OrderingTable, TableFilteringTask, WithProgress, TaskStatus, WithFilter, WithFilterEpoch, ComputationStateVersion, ResultSearchState, createResultSearchState, DataSearchTable, DataSearchTask } from './computation_types.js';
5 : import { VariantKind } from '../utils/variant.js';
6 : import { CrossFilters } from './cross_filters.js';
7 : import { DataFrame, DataFrameRegistry } from './data_frame.js';
8 : import type { EmbeddedClientContext } from '../platform/database/embedded_database.js';
9 : import { COLUMN_AGGREGATION_TASK, DATA_SEARCH_TASK, FILTERED_COLUMN_AGGREGATION_TASK, SYSTEM_COLUMN_COMPUTATION_TASK, TABLE_AGGREGATION_TASK, TABLE_FILTERING_TASK, TABLE_ORDERING_TASK, TaskVariant } from './computation_types.js';
10 : import { parseDataSearchMatches } from './sql/query_result_search_sql.js';
11 :
12 : const LOG_CTX = 'computation_state';
13 :
14 : /// The table computation state
15 : export interface TableComputationState {
16 : /// The table id
17 : tableId: number;
18 : /// The connection that ran the query of the table, its query history lists the computation queries.
19 : /// Null for a host without connections.
20 : connectionId: string | null;
21 : /// The client context of the queries of a computation without a connection, e.g. of the connection overview
22 : clientContext: EmbeddedClientContext | null;
23 : /// The tasks
24 : tasks: TableComputationTasks;
25 :
26 : /// The current version tracking data, filter, and ordering changes
27 : version: ComputationStateVersion;
28 : /// The table on the main thread
29 : dataTable: arrow.Table;
30 : /// The table field index
31 : dataTableFieldsByName: Map<string, number>;
32 : /// The abort controller
33 : dataTableLifetime: AbortController;
34 : /// The data frame in the compute module
35 : dataFrame: DataFrame | null;
36 : /// The grid columns
37 : columnGroups: ColumnGroup[];
38 : /// The ordering constraints
39 : dataTableOrdering: OrderByConstraint[];
40 :
41 : /// The active cross-filter selection.
42 : /// This is the source of truth for cross-filtering and lives here (rather than
43 : /// as component-local state) so that every view of the same table - the result
44 : /// table header plots and the visualization tab's aggregation bar - shares one
45 : /// selection and drives a single `filterTable`.
46 : crossFilters: CrossFilters;
47 : /// The active filter table (if any)
48 : filterTable: FilterTable | null;
49 : /// The active ordering table (if any)
50 : orderingTable: OrderingTable | null;
51 : /// Shared Columns search for every view of this result
52 : columnSearch: ResultSearchState;
53 : /// Shared Data search for every view of this result
54 : dataSearch: ResultSearchState;
55 : /// The materialized rows and matching columns for the applied Data search
56 : dataSearchTable: DataSearchTable | null;
57 : /// The row number column group
58 : rowNumberColumnGroup: number | null;
59 : /// The row number column name
60 : rowNumberColumnName: string | null;
61 : /// The table aggregation
62 : tableAggregation: TableAggregation | null;
63 : /// The column (group) aggregates
64 : columnAggregates: (ColumnAggregationVariant | null)[];
65 : /// The filtered column (group) aggregates
66 : filteredColumnAggregates: (WithFilterEpoch<ColumnAggregationVariant> | null)[];
67 : /// Indicates which filtered column aggregates are outdated for the current filter
68 : filteredColumnAggregatesOutdated: boolean[];
69 : }
70 :
71 : export interface TableComputationTasks {
72 : /// The table aggregation task
73 : tableAggregationTask: WithProgress<TableAggregationTask> | null;
74 : /// The task to precompute system columns
75 : systemColumnTask: WithProgress<SystemColumnComputationTask> | null;
76 : /// The filtering task
77 : filteringTask: WithProgress<TableFilteringTask> | null;
78 : /// The ordering task
79 : orderingTask: WithProgress<TableOrderingTask> | null;
80 : /// The Data-search materialization task
81 : dataSearchTask: WithProgress<DataSearchTask> | null;
82 : /// The task to compute column (group) aggregates.
83 : /// The array contains N entries where N is the number of column groups
84 : columnAggregationTasks: (WithProgress<ColumnAggregationTask> | null)[];
85 : /// The task to compute filtered column (group) aggregates
86 : filteredColumnAggregationTasks: (WithProgress<WithFilter<ColumnAggregationTask>> | null)[];
87 : }
88 :
89 : /// The computation registry.
90 : /// We store the scheduler tasks here as well to allow for linking the "latest" tasks safely in the per-table computation states.
91 : export interface ComputationState {
92 : /// The computations
93 : tableComputations: { [key: number]: TableComputationState };
94 :
95 : /// The scheduler tasks
96 : schedulerTasks: { [key: number]: TaskVariant };
97 : /// The next task id
98 : nextSchedulerTaskId: number;
99 : }
100 :
101 : /// Create the computation state
102 1 : export function createComputationState(): ComputationState {
103 1 : return {
104 1 : tableComputations: {},
105 1 : schedulerTasks: {},
106 1 : nextSchedulerTaskId: 1,
107 1 : };
108 1 : }
109 :
110 1 : export function createArrowFieldIndex(table: arrow.Table): Map<string, number> {
111 1 : let out = new Map<string, number>();
112 1 : for (let i = 0; i < table.schema.fields.length; ++i) {
113 1 : out.set(table.schema.fields[i].name, i);
114 : }
115 1 : return out;
116 1 : }
117 :
118 : /// Create the table computation state
119 1 : export function createTableComputationState(computationId: number, table: arrow.Table, tableColumns: ColumnGroup[], tableLifetime: AbortController, connectionId: string | null = null, clientContext: EmbeddedClientContext | null = null): TableComputationState {
120 1 : return {
121 1 : tableId: computationId,
122 1 : connectionId,
123 1 : clientContext,
124 1 : tasks: {
125 1 : filteringTask: null,
126 1 : orderingTask: null,
127 1 : dataSearchTask: null,
128 1 : tableAggregationTask: null,
129 1 : systemColumnTask: null,
130 1 : columnAggregationTasks: Array(tableColumns.length + 1).fill(null),
131 1 : filteredColumnAggregationTasks: Array(tableColumns.length + 1).fill(null),
132 1 : },
133 1 : dataTable: table,
134 1 : version: new ComputationStateVersion(0, 0),
135 1 : dataTableFieldsByName: createArrowFieldIndex(table),
136 1 : columnGroups: tableColumns,
137 1 : columnAggregates: Array.from({ length: tableColumns.length }, () => null),
138 1 : rowNumberColumnGroup: null,
139 1 : rowNumberColumnName: null,
140 1 : dataTableLifetime: tableLifetime,
141 1 : dataTableOrdering: [],
142 1 : crossFilters: new CrossFilters(),
143 1 : dataFrame: null,
144 1 : filterTable: null,
145 1 : orderingTable: null,
146 1 : columnSearch: createResultSearchState(),
147 1 : dataSearch: createResultSearchState(),
148 1 : dataSearchTable: null,
149 1 : filteredColumnAggregates: Array.from({ length: tableColumns.length }, () => null),
150 1 : filteredColumnAggregatesOutdated: Array.from({ length: tableColumns.length }, () => false),
151 1 : tableAggregation: null,
152 1 : };
153 1 : }
154 :
155 1 : export const COMPUTATION_FROM_QUERY_RESULT = Symbol('COMPUTATION_FROM_QUERY_RESULT');
156 1 : export const DELETE_COMPUTATION = Symbol('DELETE_COMPUTATION');
157 1 : export const CREATED_DATA_FRAME = Symbol('CREATED_DATA_FRAME');
158 1 : export const SCHEDULE_TASK = Symbol('SCHEDULE_TASK');
159 1 : export const UPDATE_SCHEDULER_TASK = Symbol('UPDATE_SCHEDULER_TASK');
160 1 : export const UNREGISTER_SCHEDULER_TASK = Symbol('UNREGISTER_SCHEDULER_TASK');
161 1 : export const TABLE_ORDERING_SUCCEDED = Symbol('TABLE_ORDERING_SUCCEDED');
162 1 : export const CLEAR_TABLE_ORDERING = Symbol('CLEAR_TABLE_ORDERING');
163 1 : export const TABLE_FILTERING_SUCCEEDED = Symbol('TABLE_FILTERING_SUCCEEDED');
164 1 : export const TABLE_AGGREGATION_SUCCEEDED = Symbol('TABLE_AGGREGATION_SUCCEEDED');
165 1 : export const SYSTEM_COLUMN_COMPUTATION_SUCCEEDED = Symbol('SYSTEM_COLUMN_COMPUTATION_SUCCEEDED');
166 1 : export const COLUMN_AGGREGATION_SUCCEEDED = Symbol('COLUMN_AGGREGATION_SUCCEEDED');
167 1 : export const FILTERED_COLUMN_AGGREGATION_SUCCEEDED = Symbol('FILTERED_COLUMN_AGGREGATION_SUCCEEDED');
168 1 : export const SET_CROSS_FILTERS = Symbol('SET_CROSS_FILTERS');
169 1 : export const SET_RESULT_SEARCH_PATTERN = Symbol('SET_RESULT_SEARCH_PATTERN');
170 1 : export const COLUMN_SEARCH_SUCCEEDED = Symbol('COLUMN_SEARCH_SUCCEEDED');
171 1 : export const DATA_SEARCH_SUCCEEDED = Symbol('DATA_SEARCH_SUCCEEDED');
172 1 : export const DATA_SEARCH_FAILED = Symbol('DATA_SEARCH_FAILED');
173 1 : export const RESULT_SEARCH_FAILED = Symbol('RESULT_SEARCH_FAILED');
174 :
175 : export type ComputationAction =
176 : | VariantKind<typeof SCHEDULE_TASK, TaskVariant>
177 : | VariantKind<typeof UPDATE_SCHEDULER_TASK, [TaskVariant, Partial<TaskProgress>]>
178 : | VariantKind<typeof UNREGISTER_SCHEDULER_TASK, TaskVariant>
179 :
180 : | VariantKind<typeof COMPUTATION_FROM_QUERY_RESULT, [number, arrow.Table, ColumnGroup[], AbortController, (string | null)?, (EmbeddedClientContext | null)?]>
181 : | VariantKind<typeof DELETE_COMPUTATION, [number]>
182 : | VariantKind<typeof CREATED_DATA_FRAME, [number, DataFrame]>
183 :
184 : | VariantKind<typeof TABLE_ORDERING_SUCCEDED, [number, OrderingTable]>
185 : | VariantKind<typeof CLEAR_TABLE_ORDERING, number>
186 : | VariantKind<typeof TABLE_FILTERING_SUCCEEDED, [number, ComputationStateVersion, FilterTable | null]>
187 : | VariantKind<typeof TABLE_AGGREGATION_SUCCEEDED, [number, TableAggregation]>
188 : | VariantKind<typeof SYSTEM_COLUMN_COMPUTATION_SUCCEEDED, [number, arrow.Table, DataFrame, ColumnGroup[]]>
189 : | VariantKind<typeof COLUMN_AGGREGATION_SUCCEEDED, [number, number, ColumnAggregationVariant]>
190 : | VariantKind<typeof FILTERED_COLUMN_AGGREGATION_SUCCEEDED, [number, number, ComputationStateVersion, WithFilterEpoch<ColumnAggregationVariant> | null]>
191 : | VariantKind<typeof SET_CROSS_FILTERS, [number, CrossFilters]>
192 : | VariantKind<typeof SET_RESULT_SEARCH_PATTERN, [number, 'columns' | 'data', string]>
193 : | VariantKind<typeof COLUMN_SEARCH_SUCCEEDED, [number, DataFrame, number, number[]]>
194 : | VariantKind<typeof DATA_SEARCH_SUCCEEDED, [number, DataFrame, number, DataSearchTable]>
195 : | VariantKind<typeof DATA_SEARCH_FAILED, [number, DataFrame, number, string]>
196 : | VariantKind<typeof RESULT_SEARCH_FAILED, [number, DataFrame, 'columns' | 'data', number, string]>
197 : ;
198 :
199 1 : export function reduceComputationState(state: ComputationState, action: ComputationAction, memory: DataFrameRegistry): ComputationState {
200 1 : switch (action.type) {
201 1 : case SCHEDULE_TASK: {
202 1 : const taskVariant = action.value;
203 : const initialProgress: TaskProgress = {
204 1 : status: TaskStatus.TASK_RUNNING,
205 1 : startedAt: new Date(),
206 1 : completedAt: null,
207 1 : failedAt: null,
208 1 : failedWithError: null,
209 : };
210 1 : return updateTask(state, taskVariant, initialProgress, memory, true);
211 1 : }
212 1 : case UPDATE_SCHEDULER_TASK: {
213 1 : const [taskVariant, progress] = action.value;
214 1 : return updateTask(state, taskVariant, progress, memory, false);
215 1 : }
216 1 : case UNREGISTER_SCHEDULER_TASK: {
217 1 : if (action.value.taskId === undefined) {
218 0 : return state;
219 : }
220 1 : if (state.schedulerTasks[action.value.taskId] !== action.value) {
221 0 : return state;
222 : }
223 1 : const schedulerTasks = { ...state.schedulerTasks };
224 1 : delete schedulerTasks[action.value.taskId];
225 1 : releaseTaskDataFrames(action.value, memory);
226 1 : return {
227 1 : ...state,
228 1 : schedulerTasks
229 1 : };
230 1 : }
231 :
232 1 : case COMPUTATION_FROM_QUERY_RESULT: {
233 1 : const [tableId, table, tableColumns, tableLifetime, connectionId, clientContext] = action.value;
234 1 : const previousTableState = state.tableComputations[tableId];
235 1 : if (previousTableState !== undefined) {
236 1 : destroyTableComputationState(previousTableState, memory);
237 : }
238 1 : const tableState = createTableComputationState(tableId, table, tableColumns, tableLifetime, connectionId ?? null, clientContext ?? null);
239 1 : return {
240 1 : ...state,
241 1 : tableComputations: {
242 1 : ...state.tableComputations,
243 1 : [tableId]: tableState
244 1 : }
245 1 : };
246 1 : }
247 1 : case DELETE_COMPUTATION: {
248 1 : const [tableId] = action.value;
249 1 : const tableState = state.tableComputations[tableId];
250 1 : if (tableState === undefined) {
251 1 : return state;
252 : }
253 1 : const tableComputations = { ...state.tableComputations };
254 1 : destroyTableComputationState(tableState, memory);
255 1 : delete tableComputations[tableId];
256 1 : return {
257 1 : ...state,
258 1 : tableComputations
259 1 : };
260 1 : }
261 1 : case CREATED_DATA_FRAME: {
262 1 : const [tableId, dataFrame] = action.value;
263 1 : memory.acquire(dataFrame);
264 :
265 1 : const prevTableState = state.tableComputations[tableId]!;
266 1 : const nextTableState: TableComputationState = {
267 1 : ...prevTableState,
268 1 : version: prevTableState.version.withDataIncrement(),
269 1 : dataFrame,
270 1 : };
271 1 : memory.release(prevTableState.dataFrame);
272 1 : return {
273 1 : ...state,
274 1 : tableComputations: {
275 1 : ...state.tableComputations,
276 1 : [tableId]: nextTableState
277 1 : }
278 1 : };
279 1 : }
280 1 : case TABLE_ORDERING_SUCCEDED: {
281 1 : const [tableId, orderingTable] = action.value;
282 1 : memory.acquire(orderingTable.dataFrame);
283 :
284 1 : const tableState = state.tableComputations[tableId];
285 1 : if (tableState === undefined) {
286 0 : memory.release(orderingTable.dataFrame);
287 0 : return state;
288 0 : }
289 : // Ordering depends on both data and filter, so check both
290 1 : if (!orderingTable.version.filterMatches(tableState.version)) {
291 1 : memory.release(orderingTable.dataFrame);
292 1 : return state;
293 1 : }
294 1 : if ((tableState.dataSearchTable?.requestId ?? null) !== (orderingTable.dataSearchRequestId ?? null)) {
295 0 : memory.release(orderingTable.dataFrame);
296 0 : return state;
297 0 : }
298 1 : const requestedOrdering = tableState.tasks.orderingTask?.orderingConstraints;
299 1 : if (requestedOrdering == null || !areOrderingConstraintsEqual(requestedOrdering, orderingTable.orderingConstraints)) {
300 1 : memory.release(orderingTable.dataFrame);
301 1 : return state;
302 1 : }
303 :
304 1 : const task = !tableState.tasks.orderingTask ? null : {
305 1 : ...tableState.tasks.orderingTask,
306 1 : progress: {
307 1 : ...tableState.tasks.orderingTask.progress,
308 1 : status: TaskStatus.TASK_SUCCEEDED,
309 1 : completedAt: new Date(),
310 1 : }
311 1 : };
312 1 : memory.release(tableState.orderingTable?.dataFrame);
313 1 : return {
314 1 : ...state,
315 1 : tableComputations: {
316 1 : ...state.tableComputations,
317 1 : [tableId]: {
318 1 : ...tableState,
319 1 : orderingTable,
320 1 : dataTableOrdering: orderingTable.orderingConstraints,
321 1 : tasks: {
322 1 : ...tableState.tasks,
323 1 : orderingTask: task
324 1 : }
325 1 : }
326 1 : }
327 1 : };
328 1 : }
329 1 : case CLEAR_TABLE_ORDERING: {
330 1 : const tableState = state.tableComputations[action.value];
331 1 : if (tableState === undefined) {
332 0 : return state;
333 : }
334 1 : memory.release(tableState.orderingTable?.dataFrame);
335 1 : return {
336 1 : ...state,
337 1 : tableComputations: {
338 1 : ...state.tableComputations,
339 1 : [action.value]: {
340 1 : ...tableState,
341 1 : dataTableOrdering: [],
342 1 : orderingTable: null,
343 1 : tasks: {
344 1 : ...tableState.tasks,
345 1 : orderingTask: tableState.tasks.orderingTask == null ? null : {
346 0 : ...tableState.tasks.orderingTask,
347 0 : orderingConstraints: [],
348 0 : },
349 1 : },
350 1 : },
351 1 : },
352 1 : };
353 1 : }
354 1 : case TABLE_FILTERING_SUCCEEDED: {
355 1 : const [tableId, completedVersion, filterTable] = action.value;
356 1 : memory.acquire(filterTable?.dataFrame);
357 1 : const tableState = state.tableComputations[tableId];
358 1 : if (tableState === undefined) {
359 0 : memory.release(filterTable?.dataFrame);
360 0 : return state;
361 0 : }
362 : // Reject stale filter results (must match current state exactly)
363 : // With pre-allocated versions, only accept results that match the current target
364 1 : if (!completedVersion.filterMatches(tableState.version)) {
365 1 : memory.release(filterTable?.dataFrame);
366 1 : return state;
367 1 : }
368 1 : return tableFilteringSucceded(state, tableId, filterTable, memory);
369 1 : }
370 1 : case TABLE_AGGREGATION_SUCCEEDED: {
371 1 : const [tableId, tableAggregation] = action.value;
372 1 : memory.acquire(tableAggregation.dataFrame);
373 :
374 1 : const tableState = state.tableComputations[tableId];
375 1 : if (tableState === undefined) {
376 0 : memory.release(tableAggregation.dataFrame);
377 0 : return state;
378 0 : }
379 1 : memory.release(tableState.tableAggregation?.dataFrame);
380 :
381 1 : const task = !tableState.tasks.tableAggregationTask ? null : {
382 1 : ...tableState.tasks.tableAggregationTask,
383 1 : progress: {
384 1 : ...tableState.tasks.tableAggregationTask.progress,
385 1 : status: TaskStatus.TASK_SUCCEEDED,
386 1 : completedAt: new Date(),
387 1 : }
388 1 : };
389 1 : return {
390 1 : ...state,
391 1 : tableComputations: {
392 1 : ...state.tableComputations,
393 1 : [tableId]: {
394 1 : ...tableState,
395 1 : tableAggregation: tableAggregation,
396 1 : tasks: {
397 1 : ...tableState.tasks,
398 1 : tableAggregationTask: task,
399 1 : }
400 1 : }
401 1 : }
402 1 : };
403 1 : }
404 1 : case SYSTEM_COLUMN_COMPUTATION_SUCCEEDED: {
405 1 : const [tableId, dataTable, dataFrame, columnGroups] = action.value;
406 1 : memory.acquire(dataFrame);
407 :
408 1 : let rowNumberColumnGroup: number | null = null;
409 1 : let rowNumberColumnName: string | null = null;
410 1 : for (let i = 0; i < columnGroups.length; ++i) {
411 1 : const group = columnGroups[i];
412 1 : switch (group.type) {
413 1 : case ROWNUMBER_COLUMN:
414 1 : rowNumberColumnGroup = i;
415 1 : rowNumberColumnName = group.value.rowNumberFieldName;
416 : break;
417 1 : }
418 1 : }
419 1 : const tableState = state.tableComputations[tableId];
420 1 : if (tableState === undefined) {
421 0 : memory.release(dataFrame);
422 0 : return state;
423 0 : }
424 1 : memory.release(tableState.dataFrame);
425 :
426 1 : const task = !tableState.tasks.systemColumnTask ? null : {
427 1 : ...tableState.tasks.systemColumnTask,
428 1 : progress: {
429 1 : ...tableState.tasks.systemColumnTask.progress,
430 1 : status: TaskStatus.TASK_SUCCEEDED,
431 1 : completedAt: new Date(),
432 1 : }
433 1 : };
434 1 : return {
435 1 : ...state,
436 1 : tableComputations: {
437 1 : ...state.tableComputations,
438 1 : [tableId]: {
439 1 : ...tableState,
440 1 : version: tableState.version.withDataIncrement(),
441 1 : dataTable,
442 1 : dataFrame,
443 1 : columnGroups,
444 1 : rowNumberColumnGroup,
445 1 : rowNumberColumnName,
446 1 : tasks: {
447 1 : ...tableState.tasks,
448 1 : systemColumnTask: task,
449 1 : }
450 1 : }
451 1 : }
452 1 : };
453 1 : }
454 1 : case COLUMN_AGGREGATION_SUCCEEDED: {
455 1 : const [tableId, columnId, columnAggregate] = action.value;
456 1 : acquireColumnAggregate(columnAggregate, memory);
457 :
458 1 : const tableState = state.tableComputations[tableId];
459 1 : if (tableState === undefined) {
460 1 : releaseColumnAggregate(columnAggregate, memory);
461 1 : return state;
462 1 : }
463 : // Column aggregation unknown?
464 1 : const currentTask = tableState.tasks.columnAggregationTasks[columnId];
465 1 : if (currentTask == null) {
466 1 : releaseColumnAggregate(columnAggregate, memory);
467 1 : return state;
468 1 : }
469 : // Destroy the previous column aggregate, if there is one
470 1 : const prevAggregate = tableState.columnAggregates[columnId];
471 1 : if (prevAggregate != null) {
472 1 : releaseColumnAggregate(prevAggregate, memory);
473 : }
474 1 : const columnAggregates: (ColumnAggregationVariant | null)[] = [...tableState.columnAggregates];
475 1 : const columnAggregationTasks = [...tableState.tasks.columnAggregationTasks];
476 1 : columnAggregates[columnId] = columnAggregate;
477 : const aggregationTask: WithProgress<ColumnAggregationTask> = {
478 1 : ...currentTask,
479 1 : progress: {
480 1 : ...currentTask.progress,
481 1 : completedAt: new Date(),
482 1 : status: TaskStatus.TASK_SUCCEEDED,
483 1 : }
484 : };
485 1 : columnAggregationTasks[columnId] = aggregationTask;
486 :
487 1 : return {
488 1 : ...state,
489 1 : tableComputations: {
490 1 : ...state.tableComputations,
491 1 : [tableId]: {
492 1 : ...tableState,
493 1 : columnAggregates,
494 1 : tasks: {
495 1 : ...tableState.tasks,
496 1 : columnAggregationTasks
497 1 : }
498 1 : }
499 1 : }
500 1 : };
501 1 : }
502 1 : case FILTERED_COLUMN_AGGREGATION_SUCCEEDED: {
503 1 : const [tableId, columnId, completedFilterVersion, newColumnAggregate] = action.value;
504 1 : acquireColumnAggregate(newColumnAggregate, memory);
505 :
506 : // Computation not registered?
507 1 : const tableState = state.tableComputations[tableId];
508 1 : if (tableState === undefined) {
509 0 : releaseColumnAggregate(newColumnAggregate, memory);
510 0 : return state;
511 0 : }
512 : // Column aggregation task unknown?
513 1 : const currentTask = tableState.tasks.filteredColumnAggregationTasks[columnId];
514 1 : if (currentTask == null) {
515 0 : releaseColumnAggregate(newColumnAggregate, memory);
516 0 : return state;
517 0 : }
518 1 : if (!currentTask.tableVersion.filterMatches(completedFilterVersion)) {
519 1 : releaseColumnAggregate(newColumnAggregate, memory);
520 1 : return state;
521 1 : }
522 1 : const currentSelectionVersion = tableState.filterTable?.version ?? tableState.version;
523 1 : if (!currentSelectionVersion.filterMatches(completedFilterVersion)) {
524 1 : releaseColumnAggregate(newColumnAggregate, memory);
525 1 : return state;
526 1 : }
527 1 : if ((newColumnAggregate?.dataSearchRequestId ?? null) !== (tableState.dataSearchTable?.requestId ?? null)) {
528 0 : releaseColumnAggregate(newColumnAggregate, memory);
529 0 : return state;
530 0 : }
531 1 : if ((currentTask.dataSearchTable?.requestId ?? null) !== (tableState.dataSearchTable?.requestId ?? null)) {
532 0 : releaseColumnAggregate(newColumnAggregate, memory);
533 0 : return state;
534 0 : }
535 :
536 1 : const filteredColumnAggregates: (WithFilterEpoch<ColumnAggregationVariant> | null)[] = [...tableState.filteredColumnAggregates];
537 1 : const filteredColumnAggregatesOutdated = [...tableState.filteredColumnAggregatesOutdated];
538 1 : const filteredColumnAggregationTasks = [...tableState.tasks.filteredColumnAggregationTasks];
539 1 : const prevColumnAggregate = filteredColumnAggregates[columnId];
540 1 : filteredColumnAggregates[columnId] = newColumnAggregate;
541 : const task: WithProgress<WithFilter<ColumnAggregationTask>> = {
542 1 : ...currentTask,
543 1 : progress: {
544 1 : ...currentTask.progress,
545 1 : completedAt: new Date(),
546 1 : status: TaskStatus.TASK_SUCCEEDED,
547 1 : }
548 : };
549 1 : filteredColumnAggregationTasks[columnId] = task;
550 :
551 1 : const aggregateUpToDate = newColumnAggregate != null
552 1 : && newColumnAggregate.filterVersion.filterMatches(currentSelectionVersion)
553 1 : && newColumnAggregate.dataSearchRequestId === (tableState.dataSearchTable?.requestId ?? null);
554 1 : filteredColumnAggregatesOutdated[columnId] = !aggregateUpToDate;
555 :
556 : // Destroy the previous column aggregate, if there is one
557 1 : if (prevColumnAggregate != null) {
558 0 : releaseColumnAggregate(prevColumnAggregate, memory);
559 : }
560 1 : return {
561 1 : ...state,
562 1 : tableComputations: {
563 1 : ...state.tableComputations,
564 1 : [tableId]: {
565 1 : ...tableState,
566 1 : filteredColumnAggregates,
567 1 : filteredColumnAggregatesOutdated,
568 1 : tasks: {
569 1 : ...tableState.tasks,
570 1 : filteredColumnAggregationTasks
571 1 : }
572 1 : }
573 1 : },
574 1 : };
575 1 : }
576 1 : case SET_CROSS_FILTERS: {
577 1 : const [tableId, crossFilters] = action.value;
578 1 : const tableState = state.tableComputations[tableId];
579 1 : if (tableState === undefined) {
580 1 : return state;
581 : }
582 : // No change? Keep the same state to avoid a spurious re-render / filtering pass.
583 1 : if (tableState.crossFilters.equals(crossFilters)) {
584 1 : return state;
585 : }
586 1 : return {
587 1 : ...state,
588 1 : tableComputations: {
589 1 : ...state.tableComputations,
590 1 : [tableId]: {
591 1 : ...tableState,
592 1 : crossFilters,
593 1 : }
594 1 : }
595 1 : };
596 1 : }
597 1 : case SET_RESULT_SEARCH_PATTERN: {
598 1 : const [tableId, kind, pattern] = action.value;
599 1 : const tableState = state.tableComputations[tableId];
600 1 : if (tableState === undefined) {
601 0 : return state;
602 : }
603 1 : const current = kind === 'columns' ? tableState.columnSearch : tableState.dataSearch;
604 1 : if (current.requestedPattern === pattern && current.pending === (pattern.length > 0)) {
605 0 : return state;
606 : }
607 1 : const nextSearch: ResultSearchState = pattern.length === 0
608 0 : ? {
609 0 : ...createResultSearchState(),
610 0 : requestId: current.requestId + 1,
611 0 : appliedRequestId: current.requestId + 1,
612 0 : }
613 1 : : {
614 1 : ...current,
615 1 : requestedPattern: pattern,
616 1 : requestId: current.requestId + 1,
617 1 : pending: true,
618 1 : error: null,
619 1 : };
620 1 : if (kind === 'data') {
621 1 : tableState.tasks.dataSearchTask?.abortController.abort();
622 : }
623 1 : const clearAppliedDataSearch = kind === 'data' && pattern.length === 0;
624 1 : if (clearAppliedDataSearch) {
625 0 : memory.release(tableState.dataSearchTable?.dataFrame);
626 0 : memory.release(tableState.orderingTable?.dataFrame);
627 0 : memory.release(tableState.tasks.dataSearchTask?.inputDataFrame);
628 0 : if (tableState.filterTable == null) {
629 0 : releaseColumnAggregates(tableState.filteredColumnAggregates, memory);
630 : }
631 0 : }
632 1 : const stateForOutdatedCheck = clearAppliedDataSearch
633 0 : ? { ...tableState, dataSearchTable: null }
634 1 : : tableState;
635 1 : return {
636 1 : ...state,
637 1 : tableComputations: {
638 1 : ...state.tableComputations,
639 1 : [tableId]: {
640 1 : ...tableState,
641 1 : columnSearch: kind === 'columns' ? nextSearch : tableState.columnSearch,
642 1 : dataSearch: kind === 'data' ? nextSearch : tableState.dataSearch,
643 1 : dataSearchTable: clearAppliedDataSearch ? null : tableState.dataSearchTable,
644 1 : orderingTable: clearAppliedDataSearch ? null : tableState.orderingTable,
645 1 : tasks: clearAppliedDataSearch
646 0 : ? { ...tableState.tasks, dataSearchTask: null }
647 1 : : tableState.tasks,
648 1 : filteredColumnAggregatesOutdated: clearAppliedDataSearch
649 0 : ? getFilteredColumnAggregatesOutdated(stateForOutdatedCheck, null)
650 1 : : tableState.filteredColumnAggregatesOutdated,
651 1 : filteredColumnAggregates: clearAppliedDataSearch && tableState.filterTable == null
652 0 : ? Array.from({ length: tableState.filteredColumnAggregates.length }, () => null)
653 1 : : tableState.filteredColumnAggregates,
654 1 : }
655 1 : }
656 1 : };
657 1 : }
658 1 : case COLUMN_SEARCH_SUCCEEDED: {
659 1 : const [tableId, sourceDataFrame, requestId, matchingColumnGroups] = action.value;
660 1 : const tableState = state.tableComputations[tableId];
661 1 : if (tableState === undefined || tableState.dataFrame !== sourceDataFrame || tableState.columnSearch.requestId !== requestId) {
662 0 : return state;
663 : }
664 1 : return {
665 1 : ...state,
666 1 : tableComputations: {
667 1 : ...state.tableComputations,
668 1 : [tableId]: {
669 1 : ...tableState,
670 1 : columnSearch: {
671 1 : ...tableState.columnSearch,
672 1 : appliedPattern: tableState.columnSearch.requestedPattern,
673 1 : appliedRequestId: requestId,
674 1 : matchingColumnGroups,
675 1 : pending: false,
676 1 : error: null,
677 1 : }
678 1 : }
679 1 : }
680 1 : };
681 1 : }
682 1 : case DATA_SEARCH_SUCCEEDED: {
683 1 : const [tableId, sourceDataFrame, requestId, dataSearchTable] = action.value;
684 1 : memory.acquire(dataSearchTable.dataFrame);
685 1 : const tableState = state.tableComputations[tableId];
686 1 : if (
687 1 : tableState === undefined
688 1 : || tableState.dataFrame !== sourceDataFrame
689 1 : || tableState.dataSearch.requestId !== requestId
690 1 : || !dataSearchTable.version.dataMatches(tableState.version)
691 0 : ) {
692 0 : memory.release(dataSearchTable.dataFrame);
693 0 : return state;
694 0 : }
695 1 : const matchingRows = parseDataSearchMatches(dataSearchTable.dataTable);
696 1 : memory.release(tableState.dataSearchTable?.dataFrame);
697 1 : memory.release(tableState.orderingTable?.dataFrame);
698 1 : if (tableState.filterTable == null) {
699 1 : releaseColumnAggregates(tableState.filteredColumnAggregates, memory);
700 : }
701 1 : const dataSearchTask = tableState.tasks.dataSearchTask == null ? null : {
702 0 : ...tableState.tasks.dataSearchTask,
703 0 : progress: {
704 0 : ...tableState.tasks.dataSearchTask.progress,
705 0 : status: TaskStatus.TASK_SUCCEEDED,
706 0 : completedAt: new Date(),
707 0 : },
708 0 : };
709 1 : const effectiveVersion = tableState.filterTable?.version ?? tableState.version;
710 1 : return {
711 1 : ...state,
712 1 : tableComputations: {
713 1 : ...state.tableComputations,
714 1 : [tableId]: {
715 1 : ...tableState,
716 1 : dataSearch: {
717 1 : ...tableState.dataSearch,
718 1 : appliedPattern: tableState.dataSearch.requestedPattern,
719 1 : appliedRequestId: requestId,
720 1 : matchingRows,
721 1 : pending: false,
722 1 : error: null,
723 1 : },
724 1 : dataSearchTable,
725 1 : orderingTable: null,
726 1 : filteredColumnAggregatesOutdated: getFilteredColumnAggregatesOutdated(
727 1 : { ...tableState, version: effectiveVersion },
728 1 : dataSearchTable,
729 1 : ),
730 1 : tasks: { ...tableState.tasks, dataSearchTask },
731 1 : }
732 1 : }
733 1 : };
734 1 : }
735 1 : case DATA_SEARCH_FAILED: {
736 1 : const [tableId, sourceDataFrame, requestId, error] = action.value;
737 1 : const tableState = state.tableComputations[tableId];
738 1 : if (tableState === undefined || tableState.dataFrame !== sourceDataFrame || tableState.dataSearch.requestId !== requestId) {
739 1 : return state;
740 : }
741 1 : memory.release(tableState.dataSearchTable?.dataFrame);
742 1 : memory.release(tableState.orderingTable?.dataFrame);
743 1 : return {
744 1 : ...state,
745 1 : tableComputations: {
746 1 : ...state.tableComputations,
747 1 : [tableId]: {
748 1 : ...tableState,
749 1 : dataSearch: {
750 1 : ...tableState.dataSearch,
751 1 : appliedPattern: '',
752 1 : appliedRequestId: requestId,
753 1 : matchingRows: null,
754 1 : pending: false,
755 1 : error,
756 1 : },
757 1 : dataSearchTable: null,
758 1 : orderingTable: null,
759 1 : filteredColumnAggregatesOutdated: getFilteredColumnAggregatesOutdated({ ...tableState, dataSearchTable: null }, null),
760 1 : filteredColumnAggregates: tableState.filterTable == null
761 1 : ? Array.from({ length: tableState.filteredColumnAggregates.length }, () => null)
762 0 : : tableState.filteredColumnAggregates,
763 1 : },
764 1 : },
765 1 : };
766 1 : }
767 1 : case RESULT_SEARCH_FAILED: {
768 1 : const [tableId, sourceDataFrame, kind, requestId, error] = action.value;
769 1 : const tableState = state.tableComputations[tableId];
770 1 : if (tableState === undefined || tableState.dataFrame !== sourceDataFrame) {
771 0 : return state;
772 : }
773 1 : const current = kind === 'columns' ? tableState.columnSearch : tableState.dataSearch;
774 1 : if (current.requestId !== requestId) {
775 1 : return state;
776 : }
777 1 : const nextSearch: ResultSearchState = {
778 1 : ...current,
779 1 : appliedPattern: '',
780 1 : appliedRequestId: requestId,
781 1 : matchingColumnGroups: null,
782 1 : matchingRows: null,
783 1 : pending: false,
784 1 : error,
785 1 : };
786 1 : return {
787 1 : ...state,
788 1 : tableComputations: {
789 1 : ...state.tableComputations,
790 1 : [tableId]: {
791 1 : ...tableState,
792 1 : columnSearch: kind === 'columns' ? nextSearch : tableState.columnSearch,
793 1 : dataSearch: kind === 'data' ? nextSearch : tableState.dataSearch,
794 1 : }
795 1 : }
796 1 : };
797 1 : }
798 1 : }
799 1 : }
800 :
801 1 : function updateTask(state: ComputationState, task: TaskVariant, progress: Partial<TaskProgress>, memory: DataFrameRegistry, create: boolean = false) {
802 1 : const allocatedTaskId = task.taskId !== undefined;
803 1 : if (!allocatedTaskId) {
804 1 : task = {
805 1 : ...task,
806 1 : taskId: state.nextSchedulerTaskId,
807 1 : };
808 : }
809 1 : const tableState = state.tableComputations[task.value.tableId];
810 1 : if (tableState === undefined) {
811 0 : return state;
812 : }
813 :
814 : // Note that we acquire data frames times twice for scheduler task and table computation state
815 1 : let registerTask: (tasks: TableComputationTasks) => TableComputationTasks = (t: TableComputationTasks) => t;
816 1 : let newStateVersion = tableState.version;
817 1 : switch (task.type) {
818 1 : case TABLE_FILTERING_TASK:
819 1 : if (create) {
820 1 : memory.acquire(task.value.inputDataFrame, 2);
821 : // Increment filter version when scheduling (not when accepting result)
822 : // This ensures concurrent filter tasks get unique target versions
823 1 : newStateVersion = tableState.version.withFilterIncrement();
824 : // Update the task itself with the new target version
825 1 : task = {
826 1 : ...task,
827 1 : value: {
828 1 : ...task.value,
829 1 : tableVersion: newStateVersion,
830 1 : }
831 1 : };
832 1 : }
833 1 : registerTask = (tasks: TableComputationTasks) => {
834 1 : const filterTask = task.value as TableFilteringTask;
835 1 : return ({
836 1 : ...tasks,
837 1 : filteringTask: {
838 1 : tableId: filterTask.tableId,
839 1 : tableVersion: newStateVersion,
840 1 : inputDataTable: filterTask.inputDataTable,
841 1 : inputDataTableFieldIndex: filterTask.inputDataTableFieldIndex,
842 1 : inputDataFrame: filterTask.inputDataFrame,
843 1 : rowNumberColumnName: filterTask.rowNumberColumnName,
844 1 : filters: filterTask.filters,
845 1 : progress: {
846 1 : ...tasks.filteringTask?.progress,
847 1 : ...progress,
848 1 : } as TaskProgress,
849 1 : }
850 1 : });
851 1 : };
852 1 : break;
853 1 : case TABLE_ORDERING_TASK:
854 1 : if (create) {
855 1 : memory.release(tableState.tasks.orderingTask?.inputDataFrame);
856 1 : memory.release(tableState.tasks.orderingTask?.filterTable?.dataFrame);
857 1 : memory.release(tableState.tasks.orderingTask?.dataSearchTable?.dataFrame);
858 1 : memory.acquire(task.value.inputDataFrame, 2);
859 1 : memory.acquire(task.value.filterTable?.dataFrame, 2);
860 1 : memory.acquire(task.value.dataSearchTable?.dataFrame, 2);
861 1 : }
862 1 : registerTask = (tasks: TableComputationTasks) => {
863 1 : const orderTask = task.value as TableOrderingTask;
864 1 : if (
865 1 : !create
866 1 : && tasks.orderingTask != null
867 1 : && !areOrderingConstraintsEqual(tasks.orderingTask.orderingConstraints, orderTask.orderingConstraints)
868 : ) {
869 0 : return tasks;
870 : }
871 1 : return ({
872 1 : ...tasks,
873 1 : orderingTask: {
874 1 : tableId: orderTask.tableId,
875 1 : tableVersion: tableState.version,
876 1 : inputDataTable: orderTask.inputDataTable,
877 1 : inputDataTableFieldIndex: orderTask.inputDataTableFieldIndex,
878 1 : inputDataFrame: orderTask.inputDataFrame,
879 1 : filterTable: orderTask.filterTable,
880 1 : dataSearchTable: orderTask.dataSearchTable,
881 1 : rowNumberColumnName: orderTask.rowNumberColumnName,
882 1 : orderingConstraints: orderTask.orderingConstraints,
883 1 : progress: {
884 1 : ...tasks.orderingTask?.progress,
885 1 : ...progress,
886 1 : } as TaskProgress,
887 1 : }
888 1 : });
889 1 : };
890 1 : break;
891 1 : case DATA_SEARCH_TASK:
892 1 : if (create) {
893 1 : tableState.tasks.dataSearchTask?.abortController.abort();
894 1 : memory.release(tableState.tasks.dataSearchTask?.inputDataFrame);
895 1 : memory.acquire(task.value.inputDataFrame, 2);
896 1 : }
897 1 : registerTask = (tasks: TableComputationTasks) => {
898 1 : const searchTask = task.value as DataSearchTask;
899 1 : if (!create && tasks.dataSearchTask?.requestId !== searchTask.requestId) {
900 1 : return tasks;
901 : }
902 1 : return {
903 1 : ...tasks,
904 1 : dataSearchTask: {
905 1 : ...searchTask,
906 1 : progress: {
907 1 : ...tasks.dataSearchTask?.progress,
908 1 : ...progress,
909 1 : } as TaskProgress,
910 1 : },
911 1 : };
912 1 : };
913 1 : break;
914 1 : case TABLE_AGGREGATION_TASK:
915 1 : if (create) {
916 1 : memory.acquire(task.value.inputDataFrame, 2);
917 : }
918 1 : registerTask = (tasks: TableComputationTasks) => {
919 1 : const aggTask = task.value as TableAggregationTask;
920 1 : return ({
921 1 : ...tasks,
922 1 : tableAggregationTask: {
923 1 : tableId: aggTask.tableId,
924 1 : tableVersion: aggTask.tableVersion,
925 1 : columnEntries: aggTask.columnEntries,
926 1 : inputDataFrame: aggTask.inputDataFrame,
927 1 : progress: {
928 1 : ...tasks.tableAggregationTask?.progress,
929 1 : ...progress,
930 1 : } as TaskProgress,
931 1 : }
932 1 : });
933 1 : };
934 1 : break;
935 1 : case SYSTEM_COLUMN_COMPUTATION_TASK:
936 1 : if (create) {
937 1 : memory.acquire(task.value.inputDataFrame, 2);
938 1 : memory.acquire(task.value.tableAggregate.dataFrame, 2);
939 1 : }
940 1 : registerTask = (tasks: TableComputationTasks) => {
941 1 : const sysTask = task.value as SystemColumnComputationTask;
942 1 : return ({
943 1 : ...tasks,
944 1 : systemColumnTask: {
945 1 : tableId: sysTask.tableId,
946 1 : tableVersion: sysTask.tableVersion,
947 1 : columnEntries: sysTask.columnEntries,
948 1 : inputTable: sysTask.inputTable,
949 1 : inputDataFrame: sysTask.inputDataFrame,
950 1 : rowNumberFieldName: sysTask.rowNumberFieldName,
951 1 : tableAggregate: sysTask.tableAggregate,
952 1 : progress: {
953 1 : ...tasks.systemColumnTask?.progress,
954 1 : ...progress,
955 1 : } as TaskProgress,
956 1 : }
957 1 : });
958 1 : };
959 1 : break;
960 1 : case COLUMN_AGGREGATION_TASK:
961 1 : if (create) {
962 1 : memory.acquire(task.value.inputDataFrame, 2);
963 1 : memory.acquire(task.value.tableAggregate.dataFrame, 2);
964 1 : }
965 1 : registerTask = (tasks: TableComputationTasks) => {
966 1 : const colTask = task.value as ColumnAggregationTask;
967 1 : const prev = tasks.columnAggregationTasks[colTask.columnId];
968 1 : const updated = [...tasks.columnAggregationTasks];
969 1 : updated[colTask.columnId] = {
970 1 : tableId: colTask.tableId,
971 1 : tableVersion: colTask.tableVersion,
972 1 : columnId: colTask.columnId,
973 1 : columnEntry: colTask.columnEntry,
974 1 : inputDataFrame: colTask.inputDataFrame,
975 1 : tableAggregate: colTask.tableAggregate,
976 1 : progress: {
977 1 : ...prev?.progress,
978 1 : ...progress
979 1 : } as TaskProgress
980 1 : };
981 1 : return ({
982 1 : ...tasks,
983 1 : columnAggregationTasks: updated
984 1 : })
985 1 : };
986 1 : break;
987 1 : case FILTERED_COLUMN_AGGREGATION_TASK:
988 1 : if (create) {
989 1 : const previous = tableState.tasks.filteredColumnAggregationTasks[task.value.columnId];
990 1 : memory.release(previous?.inputDataFrame);
991 1 : memory.release(previous?.filterTable?.dataFrame);
992 1 : memory.release(previous?.dataSearchTable?.dataFrame);
993 1 : releaseColumnAggregate(previous?.unfilteredAggregate, memory);
994 1 : memory.acquire(task.value.inputDataFrame, 2);
995 1 : memory.acquire(task.value.filterTable?.dataFrame, 2);
996 1 : memory.acquire(task.value.dataSearchTable?.dataFrame, 2);
997 1 : memory.acquire(task.value.tableAggregate.dataFrame, 2);
998 1 : acquireColumnAggregate(task.value.unfilteredAggregate, memory, 2);
999 1 : }
1000 1 : registerTask = (tasks: TableComputationTasks) => {
1001 1 : const filteredTask = task.value as WithFilter<ColumnAggregationTask>;
1002 1 : const prev = tasks.filteredColumnAggregationTasks[filteredTask.columnId];
1003 1 : const updated = [...tasks.filteredColumnAggregationTasks];
1004 1 : updated[filteredTask.columnId] = {
1005 1 : tableId: filteredTask.tableId,
1006 1 : tableVersion: filteredTask.tableVersion,
1007 1 : columnId: filteredTask.columnId,
1008 1 : columnEntry: filteredTask.columnEntry,
1009 1 : inputDataFrame: filteredTask.inputDataFrame,
1010 1 : tableAggregate: filteredTask.tableAggregate,
1011 1 : filterTable: filteredTask.filterTable,
1012 1 : dataSearchTable: filteredTask.dataSearchTable,
1013 1 : selectionRowCount: filteredTask.selectionRowCount,
1014 1 : unfilteredAggregate: filteredTask.unfilteredAggregate,
1015 1 : progress: {
1016 1 : ...prev?.progress,
1017 1 : ...progress
1018 1 : } as TaskProgress
1019 1 : };
1020 1 : return ({
1021 1 : ...tasks,
1022 1 : filteredColumnAggregationTasks: updated
1023 1 : })
1024 1 : };
1025 : break;
1026 1 : }
1027 1 : const updatedTaskId = task.taskId!;
1028 : const updatedState: ComputationState = {
1029 1 : ...state,
1030 1 : nextSchedulerTaskId: (!allocatedTaskId) ? (state.nextSchedulerTaskId + 1) : state.nextSchedulerTaskId,
1031 1 : schedulerTasks: {
1032 1 : ...state.schedulerTasks,
1033 1 : [updatedTaskId]: task
1034 1 : },
1035 1 : tableComputations: {
1036 1 : ...state.tableComputations,
1037 1 : [task.value.tableId]: {
1038 1 : ...tableState,
1039 1 : version: newStateVersion,
1040 1 : tasks: registerTask(tableState.tasks)
1041 1 : }
1042 1 : }
1043 : };
1044 1 : return updatedState;
1045 1 : }
1046 :
1047 1 : function releaseTaskDataFrames(task: TaskVariant, memory: DataFrameRegistry) {
1048 1 : switch (task.type) {
1049 1 : case TABLE_FILTERING_TASK:
1050 1 : case TABLE_AGGREGATION_TASK:
1051 1 : case DATA_SEARCH_TASK:
1052 1 : memory.release(task.value.inputDataFrame);
1053 1 : break;
1054 1 : case TABLE_ORDERING_TASK:
1055 1 : memory.release(task.value.inputDataFrame);
1056 1 : memory.release(task.value.filterTable?.dataFrame);
1057 1 : memory.release(task.value.dataSearchTable?.dataFrame);
1058 1 : break;
1059 1 : case SYSTEM_COLUMN_COMPUTATION_TASK:
1060 1 : memory.release(task.value.inputDataFrame);
1061 1 : break;
1062 1 : case COLUMN_AGGREGATION_TASK:
1063 1 : memory.release(task.value.inputDataFrame);
1064 1 : break;
1065 1 : case FILTERED_COLUMN_AGGREGATION_TASK:
1066 1 : memory.release(task.value.inputDataFrame);
1067 1 : memory.release(task.value.filterTable?.dataFrame);
1068 1 : memory.release(task.value.dataSearchTable?.dataFrame);
1069 1 : releaseColumnAggregate(task.value.unfilteredAggregate, memory);
1070 : break;
1071 1 : }
1072 1 : }
1073 :
1074 1 : function areOrderingConstraintsEqual(left: OrderByConstraint[], right: OrderByConstraint[]): boolean {
1075 1 : if (left.length !== right.length) {
1076 1 : return false;
1077 : }
1078 1 : return left.every((constraint, index) => {
1079 1 : const other = right[index];
1080 1 : return constraint.field === other.field
1081 1 : && (constraint.ascending ?? true) === (other.ascending ?? true)
1082 1 : && (constraint.nullsFirst ?? false) === (other.nullsFirst ?? false);
1083 1 : });
1084 1 : }
1085 :
1086 1 : function tableFilteringSucceded(state: ComputationState, tableId: number, filterTable: FilterTable | null, memory: DataFrameRegistry) {
1087 1 : const tableState = state.tableComputations[tableId];
1088 1 : if (tableState === undefined) {
1089 0 : memory.release(filterTable?.dataFrame);
1090 0 : return state;
1091 0 : }
1092 :
1093 : // Mark filtering task succeeded
1094 1 : const filteringTask = !tableState.tasks.filteringTask ? null : {
1095 1 : ...tableState.tasks.filteringTask,
1096 1 : progress: {
1097 1 : ...tableState.tasks.filteringTask.progress,
1098 1 : status: TaskStatus.TASK_SUCCEEDED,
1099 1 : completedAt: new Date(),
1100 1 : }
1101 1 : };
1102 1 : const obsoleteFilteredColumnAggregates: Map<number, WithFilterEpoch<ColumnAggregationVariant> | null> = new Map();
1103 1 : let newFilteredColumnAggregatesOutdated = tableState.filteredColumnAggregatesOutdated;
1104 :
1105 1 : if (filterTable == null && tableState.dataSearchTable == null) {
1106 : // Clear column aggregates - track existing ones for release
1107 1 : for (let columnId = 0; columnId < tableState.filteredColumnAggregates.length; ++columnId) {
1108 1 : const prevAggregate = tableState.filteredColumnAggregates[columnId];
1109 1 : if (prevAggregate != null) {
1110 0 : obsoleteFilteredColumnAggregates.set(columnId, prevAggregate);
1111 : }
1112 1 : }
1113 1 : newFilteredColumnAggregatesOutdated = Array.from({ length: tableState.filteredColumnAggregatesOutdated.length }, () => false);
1114 :
1115 1 : } else {
1116 1 : newFilteredColumnAggregatesOutdated = getFilteredColumnAggregatesOutdated(tableState, tableState.dataSearchTable, filterTable);
1117 : }
1118 :
1119 : // When the filter is cleared, release obsolete aggregates and clear the state.
1120 : // When the filter is active, we deliberately keep old aggregates until new ones are computed.
1121 : // We use the filter epoch to determine that an aggregate is not yet updated.
1122 1 : let newFilteredColumnAggregates = tableState.filteredColumnAggregates;
1123 1 : if (filterTable == null && tableState.dataSearchTable == null) {
1124 : // Release all obsolete filtered column aggregates
1125 1 : for (const [, prevAggregate] of obsoleteFilteredColumnAggregates.entries()) {
1126 0 : releaseColumnAggregate(prevAggregate, memory);
1127 : }
1128 : // Clear the filtered column aggregates array
1129 1 : newFilteredColumnAggregates = Array.from({ length: tableState.filteredColumnAggregates.length }, () => null);
1130 1 : }
1131 :
1132 1 : memory.release(tableState.filterTable?.dataFrame);
1133 1 : memory.release(tableState.orderingTable?.dataFrame);
1134 1 : return {
1135 1 : ...state,
1136 1 : tableComputations: {
1137 1 : ...state.tableComputations,
1138 1 : [tableId]: {
1139 1 : ...tableState,
1140 : // Don't increment version here - it was already incremented when scheduling the task
1141 : // The filterTable.version should match the task's target version
1142 1 : filterTable,
1143 1 : orderingTable: null,
1144 1 : filteredColumnAggregates: newFilteredColumnAggregates,
1145 1 : filteredColumnAggregatesOutdated: newFilteredColumnAggregatesOutdated,
1146 1 : tasks: {
1147 1 : ...tableState.tasks,
1148 1 : filteringTask,
1149 1 : }
1150 1 : }
1151 1 : },
1152 1 : };
1153 1 : }
1154 :
1155 1 : function getFilteredColumnAggregatesOutdated(
1156 1 : tableState: TableComputationState,
1157 1 : dataSearchTable: DataSearchTable | null,
1158 1 : filterTable: FilterTable | null = tableState.filterTable,
1159 1 : ): boolean[] {
1160 1 : if (filterTable == null && dataSearchTable == null) {
1161 1 : return Array.from({ length: tableState.filteredColumnAggregatesOutdated.length }, () => false);
1162 : }
1163 1 : const outdated = [...tableState.filteredColumnAggregatesOutdated];
1164 1 : const targetVersion = filterTable?.version ?? tableState.version;
1165 1 : for (let columnId = 0; columnId < tableState.columnGroups.length; ++columnId) {
1166 1 : const columnEntry = tableState.columnGroups[columnId];
1167 1 : if (columnEntry.type == SKIPPED_COLUMN || columnEntry.type == ROWNUMBER_COLUMN || columnEntry.type == LIST_COLUMN) {
1168 1 : outdated[columnId] = false;
1169 1 : continue;
1170 1 : }
1171 1 : const unfilteredAggregate = tableState.columnAggregates[columnId];
1172 1 : const previous = tableState.filteredColumnAggregates[columnId];
1173 1 : outdated[columnId] = unfilteredAggregate != null && !(
1174 1 : previous?.filterVersion.filterMatches(targetVersion)
1175 0 : && previous.dataSearchRequestId === (dataSearchTable?.requestId ?? null)
1176 : );
1177 1 : }
1178 1 : return outdated;
1179 1 : }
1180 :
1181 : /// Acquire a column aggregate
1182 1 : function acquireColumnAggregate(aggregate: ColumnAggregationVariant | null | undefined, memory: DataFrameRegistry, times: number = 1) {
1183 1 : if (aggregate == null || aggregate == undefined) {
1184 0 : return;
1185 : }
1186 1 : switch (aggregate.type) {
1187 1 : case ORDINAL_COLUMN:
1188 1 : memory.acquire(aggregate.value.binnedDataFrame, times);
1189 1 : break;
1190 1 : case STRING_COLUMN:
1191 1 : memory.acquire(aggregate.value.frequentValuesDataFrame, times);
1192 1 : break;
1193 0 : case LIST_COLUMN:
1194 0 : memory.acquire(aggregate.value.frequentValuesDataFrame, times);
1195 0 : break;
1196 0 : case SKIPPED_COLUMN:
1197 : break;
1198 1 : }
1199 1 : }
1200 :
1201 : /// Helper to release memory of a column aggregate
1202 1 : function releaseColumnAggregate(aggregate: ColumnAggregationVariant | null | undefined, memory: DataFrameRegistry) {
1203 1 : if (aggregate == null || aggregate == undefined) {
1204 1 : return;
1205 : }
1206 1 : switch (aggregate.type) {
1207 1 : case ORDINAL_COLUMN:
1208 1 : memory.release(aggregate.value.binnedDataFrame);
1209 1 : break;
1210 1 : case STRING_COLUMN:
1211 1 : memory.release(aggregate.value.frequentValuesDataFrame);
1212 1 : break;
1213 0 : case LIST_COLUMN:
1214 0 : memory.release(aggregate.value.frequentValuesDataFrame);
1215 0 : break;
1216 0 : case SKIPPED_COLUMN:
1217 : break;
1218 1 : }
1219 1 : }
1220 :
1221 : /// Helper to release memory of column aggregates
1222 1 : function releaseColumnAggregates(aggregates: (ColumnAggregationVariant | null)[], dataFrameMemory: DataFrameRegistry) {
1223 1 : for (const aggregate of aggregates) {
1224 1 : if (aggregate != null) {
1225 0 : releaseColumnAggregate(aggregate, dataFrameMemory);
1226 : }
1227 : }
1228 1 : }
1229 :
1230 : /// Helper to destroy state of a table
1231 1 : function destroyTableComputationState(state: TableComputationState, memory: DataFrameRegistry) {
1232 1 : state.dataTableLifetime.abort();
1233 : // Release data frames
1234 1 : memory.release(state.dataFrame);
1235 1 : memory.release(state.filterTable?.dataFrame);
1236 1 : memory.release(state.dataSearchTable?.dataFrame);
1237 1 : memory.release(state.orderingTable?.dataFrame);
1238 1 : memory.release(state.tableAggregation?.dataFrame);
1239 1 : releaseColumnAggregates(state.columnAggregates, memory);
1240 1 : releaseColumnAggregates(state.filteredColumnAggregates, memory);
1241 :
1242 : // Release data frames from tasks
1243 1 : memory.release(state.tasks.filteringTask?.inputDataFrame);
1244 1 : memory.release(state.tasks.orderingTask?.inputDataFrame);
1245 1 : memory.release(state.tasks.orderingTask?.filterTable?.dataFrame);
1246 1 : memory.release(state.tasks.orderingTask?.dataSearchTable?.dataFrame);
1247 1 : memory.release(state.tasks.dataSearchTask?.inputDataFrame);
1248 1 : memory.release(state.tasks.tableAggregationTask?.inputDataFrame);
1249 1 : memory.release(state.tasks.systemColumnTask?.inputDataFrame);
1250 1 : memory.releaseMany(state.tasks.columnAggregationTasks.map(task => task?.inputDataFrame));
1251 1 : for (const task of state.tasks.filteredColumnAggregationTasks) {
1252 1 : memory.release(task?.filterTable?.dataFrame);
1253 1 : memory.release(task?.dataSearchTable?.dataFrame);
1254 1 : releaseColumnAggregate(task?.unfilteredAggregate ?? null, memory);
1255 1 : }
1256 1 : }
|
