Line data Source code
1 : import * as React from 'react';
2 :
3 : import { ComputationAction, SET_CROSS_FILTERS, TableComputationState } from '../../compute/computation_state.js';
4 : import { CrossFilters } from '../../compute/cross_filters.js';
5 : import { Dispatch } from '../../utils/variant.js';
6 : import { FILTERED_COLUMN_AGGREGATION_TASK, TABLE_FILTERING_TASK, ORDINAL_COLUMN, STRING_COLUMN, OrdinalColumnAggregation, StringColumnAggregation, TableAggregation, TableFilteringTask, TaskStatus, WithFilter, ColumnAggregationTask, DataSearchTable, FilterTable } from '../../compute/computation_types.js';
7 : import { ScalarFilter } from '../../compute/sql/sqlframe_builder.js';
8 : import { HistogramFilterCallback } from './histogram_cell.js';
9 : import { MostFrequentValueFilterCallback } from './mostfrequent_cell.js';
10 : import { DataTableLayout } from './data_table_layout.js';
11 : import { scheduleTask } from '../../compute/computation_tasks.js';
12 : import { forEachIntegerValue } from './visible_rows.js';
13 :
14 : /// Compare two scalar filter lists for equality.
15 1 : function scalarFiltersEqual(left: ScalarFilter[], right: ScalarFilter[]): boolean {
16 1 : if (left.length !== right.length) {
17 1 : return false;
18 : }
19 1 : for (let i = 0; i < left.length; ++i) {
20 1 : const a = left[i];
21 1 : const b = right[i];
22 1 : if (a.fieldName !== b.fieldName || a.op !== b.op || a.value !== b.value) {
23 1 : return false;
24 : }
25 1 : }
26 1 : return true;
27 1 : }
28 :
29 1 : const pendingSchedules = new WeakMap<object, Set<string>>();
30 :
31 : interface PendingFilterTask {
32 : key: string;
33 : task: TableFilteringTask;
34 : }
35 :
36 : interface FilterTaskQueue {
37 : activeKey: string | null;
38 : pending: PendingFilterTask | null;
39 : }
40 :
41 1 : const filterTaskQueues = new WeakMap<object, Map<number, FilterTaskQueue>>();
42 :
43 : /// Effects in separate consumers can run from the same state snapshot. Keep a task key
44 : /// reserved until the scheduled operation settles so only the first consumer schedules it.
45 1 : function claimSchedule(dispatch: Dispatch<ComputationAction>, key: string): (() => void) | null {
46 1 : const owner = dispatch as object;
47 1 : let schedules = pendingSchedules.get(owner);
48 1 : if (schedules == null) {
49 1 : schedules = new Set();
50 1 : pendingSchedules.set(owner, schedules);
51 1 : }
52 1 : if (schedules.has(key)) {
53 1 : return null;
54 : }
55 1 : schedules.add(key);
56 1 : return () => schedules!.delete(key);
57 1 : }
58 :
59 1 : function scalarFiltersKey(filters: ScalarFilter[]): string {
60 1 : return filters.map(filter => `${filter.fieldName}\u0000${filter.op}\u0000${filter.value}`).join('\u0001');
61 1 : }
62 :
63 : /// Run at most one filter task per table and retain only the newest request that arrives
64 : /// while it is running. Stale brush positions never enter the scheduler.
65 1 : function scheduleLatestFilter(
66 1 : dispatch: Dispatch<ComputationAction>,
67 1 : key: string,
68 1 : task: TableFilteringTask,
69 1 : ): void {
70 1 : const owner = dispatch as object;
71 1 : let queues = filterTaskQueues.get(owner);
72 1 : if (queues == null) {
73 1 : queues = new Map();
74 1 : filterTaskQueues.set(owner, queues);
75 1 : }
76 1 : let queue = queues.get(task.tableId);
77 1 : if (queue == null) {
78 1 : queue = { activeKey: null, pending: null };
79 1 : queues.set(task.tableId, queue);
80 1 : }
81 :
82 1 : if (queue.activeKey === key) {
83 0 : queue.pending = null;
84 0 : return;
85 0 : }
86 1 : if (queue.pending?.key === key) {
87 0 : return;
88 : }
89 1 : if (queue.activeKey != null) {
90 1 : queue.pending = { key, task };
91 1 : return;
92 1 : }
93 :
94 1 : const launch = (request: PendingFilterTask) => {
95 1 : queue!.activeKey = request.key;
96 1 : void scheduleTask(TABLE_FILTERING_TASK, request.task, dispatch).finally(() => {
97 1 : queue!.activeKey = null;
98 1 : const pending = queue!.pending;
99 1 : queue!.pending = null;
100 1 : if (pending != null) {
101 1 : launch(pending);
102 : } else {
103 1 : queues!.delete(request.task.tableId);
104 : }
105 1 : });
106 1 : };
107 1 : launch({ key, task });
108 1 : }
109 :
110 : export interface CrossFilterController {
111 : /// The active cross-filter selection (shared, read from the computation state)
112 : crossFilters: CrossFilters;
113 : /// Toggle/update a histogram filter for a column
114 : histogramFilter: HistogramFilterCallback;
115 : /// Toggle/update a most-frequent-value filter for a column
116 : mostFrequentValueFilter: MostFrequentValueFilterCallback;
117 : /// Request a filtered column aggregation for a column group
118 : requestFilteredColumnAggregation: (columnId: number) => void;
119 : }
120 :
121 : /// Cross-filter controller shared by every view of a table (the result table header
122 : /// plots and the visualization tab's aggregation bar).
123 : ///
124 : /// The selection lives on the shared `TableComputationState.crossFilters`, so brushing
125 : /// in any consumer updates the single selection and drives one `filterTable`. The
126 : /// filtering effect below is guarded so that mounting this hook from multiple consumers
127 : /// at once (e.g. the Data and Visualization tabs in split mode) does not schedule
128 : /// duplicate filtering tasks.
129 1 : export function useCrossFilters(
130 1 : computationState: TableComputationState | null,
131 1 : dispatchComputation: Dispatch<ComputationAction>,
132 1 : gridLayout: DataTableLayout,
133 1 : ): CrossFilterController {
134 1 : const crossFilters = computationState?.crossFilters ?? EMPTY_CROSS_FILTERS;
135 1 : const tableId = computationState?.tableId ?? null;
136 :
137 : // Keep the callbacks stable by reading layout / column groups / state through refs.
138 1 : const gridLayoutRef = React.useRef(gridLayout);
139 1 : const columnGroupsRef = React.useRef(computationState?.columnGroups ?? EMPTY_COLUMN_GROUPS);
140 1 : const crossFiltersRef = React.useRef(crossFilters);
141 1 : const stateRef = React.useRef(computationState);
142 1 : gridLayoutRef.current = gridLayout;
143 1 : columnGroupsRef.current = computationState?.columnGroups ?? EMPTY_COLUMN_GROUPS;
144 1 : crossFiltersRef.current = crossFilters;
145 1 : stateRef.current = computationState;
146 :
147 1 : const histogramFilter: HistogramFilterCallback = React.useCallback((_table: TableAggregation, columnIndex: number, _column: OrdinalColumnAggregation, brush: [number, number] | null) => {
148 1 : if (tableId == null) {
149 0 : return;
150 : }
151 1 : const columnGroupId = gridLayoutRef.current.columnGroupByColumnIndex[columnIndex];
152 1 : const columnGroup = columnGroupsRef.current[columnGroupId];
153 1 : if (columnGroup == null || columnGroup.type != ORDINAL_COLUMN) {
154 1 : return;
155 : }
156 1 : const current = crossFiltersRef.current;
157 1 : if (current.containsHistogramFilter(columnGroupId, brush)) {
158 0 : return;
159 : }
160 1 : const cloned = current.clone();
161 1 : cloned.addHistogramFilter(columnGroupId, columnGroup.value, brush);
162 1 : dispatchComputation({ type: SET_CROSS_FILTERS, value: [tableId, cloned] });
163 1 : }, [dispatchComputation, tableId]);
164 :
165 1 : const mostFrequentValueFilter: MostFrequentValueFilterCallback = React.useCallback((_table: TableAggregation, columnIndex: number, _column: StringColumnAggregation, frequentValueId: number | null) => {
166 1 : if (tableId == null) {
167 0 : return;
168 : }
169 1 : const columnGroupId = gridLayoutRef.current.columnGroupByColumnIndex[columnIndex];
170 1 : const columnGroup = columnGroupsRef.current[columnGroupId];
171 1 : if (columnGroup == null || columnGroup.type != STRING_COLUMN || columnGroup.value.valueIdFieldName == null) {
172 1 : return;
173 : }
174 1 : const current = crossFiltersRef.current;
175 1 : const nextValueId = current.containsMostFrequentValueFilter(columnGroupId, frequentValueId)
176 1 : ? null
177 1 : : frequentValueId;
178 1 : const cloned = current.clone();
179 1 : cloned.addMostFrequentValueFilter(columnGroupId, columnGroup.value, nextValueId);
180 1 : dispatchComputation({ type: SET_CROSS_FILTERS, value: [tableId, cloned] });
181 1 : }, [dispatchComputation, tableId]);
182 :
183 1 : const crossFilterTransforms = React.useMemo(
184 1 : () => crossFilters.createFilterTransforms(),
185 1 : [crossFilters],
186 1 : );
187 :
188 : // Filter the immutable base table whenever cross-filters change.
189 : // Guarded so concurrent consumers of the shared selection don't double-schedule.
190 1 : React.useEffect(() => {
191 1 : if (!computationState || !computationState.dataFrame || !computationState.rowNumberColumnName) {
192 1 : return;
193 : }
194 : // Already reflected by the current filter table?
195 1 : const currentFilterFilters = computationState.tasks.filteringTask?.filters;
196 : const filterUpToDate =
197 1 : computationState.filterTable != null
198 1 : && computationState.filterTable.version.filterMatches(computationState.version)
199 1 : && currentFilterFilters != null
200 1 : && scalarFiltersEqual(currentFilterFilters, crossFilterTransforms);
201 1 : if (filterUpToDate) {
202 1 : return;
203 : }
204 : // Empty selection with no active filter table? Nothing to do.
205 1 : if (crossFilterTransforms.length === 0 && computationState.filterTable == null) {
206 1 : return;
207 : }
208 : // A filtering task for exactly these transforms is already running?
209 1 : const runningTask = computationState.tasks.filteringTask;
210 : const hasUpToDateRunningTask =
211 1 : runningTask?.progress.status === TaskStatus.TASK_RUNNING
212 1 : && runningTask.tableVersion.filterMatches(computationState.version)
213 1 : && scalarFiltersEqual(runningTask.filters, crossFilterTransforms);
214 1 : if (hasUpToDateRunningTask) {
215 1 : return;
216 : }
217 1 : const scheduleKey = `${computationState.version.toString()}:${scalarFiltersKey(crossFilterTransforms)}`;
218 : const filteringTask: TableFilteringTask = {
219 1 : tableId: computationState.tableId,
220 1 : tableVersion: computationState.version,
221 1 : inputDataTable: computationState.dataTable,
222 1 : inputDataTableFieldIndex: computationState.dataTableFieldsByName,
223 1 : inputDataFrame: computationState.dataFrame,
224 1 : filters: crossFilterTransforms,
225 1 : rowNumberColumnName: computationState.rowNumberColumnName,
226 : };
227 1 : scheduleLatestFilter(dispatchComputation, scheduleKey, filteringTask);
228 1 : }, [
229 1 : computationState?.dataFrame,
230 1 : computationState?.dataTable,
231 1 : computationState?.dataTableFieldsByName,
232 1 : computationState?.rowNumberColumnName,
233 1 : computationState?.tableId,
234 1 : computationState?.version,
235 1 : computationState?.filterTable,
236 1 : computationState?.tasks.filteringTask,
237 1 : crossFilterTransforms,
238 1 : dispatchComputation,
239 1 : ]);
240 :
241 1 : const requestFilteredColumnAggregation = React.useCallback((columnId: number) => {
242 1 : const state = stateRef.current;
243 1 : if (state == null) {
244 0 : return;
245 : }
246 1 : const tableAggregation = state.tableAggregation;
247 1 : const filterTable = state.filterTable;
248 1 : const dataSearchTable = state.dataSearchTable;
249 1 : const inputDataFrame = state.dataFrame;
250 1 : const columnEntry = state.columnGroups[columnId];
251 1 : const unfilteredAggregate = state.columnAggregates[columnId];
252 1 : if (
253 1 : tableAggregation == null
254 1 : || (filterTable == null && dataSearchTable == null)
255 1 : || inputDataFrame == null
256 1 : || columnEntry == null
257 1 : || unfilteredAggregate == null
258 : ) {
259 1 : return;
260 : }
261 :
262 1 : const currentTask = state.tasks.filteredColumnAggregationTasks[columnId];
263 1 : const hasUpToDateRunningTask = currentTask?.progress.status === TaskStatus.TASK_RUNNING
264 0 : && currentTask.tableVersion.filterMatches(state.version)
265 0 : && (currentTask.dataSearchTable?.requestId ?? null) === (dataSearchTable?.requestId ?? null);
266 1 : if (hasUpToDateRunningTask) {
267 0 : return;
268 : }
269 :
270 1 : const scheduleKey = `aggregate:${state.tableId}:${columnId}:${state.version.toString()}:${dataSearchTable?.requestId ?? ''}`;
271 1 : const releaseSchedule = claimSchedule(dispatchComputation, scheduleKey);
272 1 : if (releaseSchedule == null) {
273 1 : return;
274 : }
275 :
276 1 : const task: WithFilter<ColumnAggregationTask> = {
277 1 : tableId: state.tableId,
278 1 : tableVersion: filterTable?.version ?? state.version,
279 1 : columnId,
280 1 : tableAggregate: tableAggregation,
281 1 : columnEntry,
282 1 : inputDataFrame,
283 1 : filterTable,
284 1 : dataSearchTable,
285 1 : selectionRowCount: countSelectionRows(filterTable, dataSearchTable),
286 1 : unfilteredAggregate,
287 1 : };
288 1 : void scheduleTask(FILTERED_COLUMN_AGGREGATION_TASK, task, dispatchComputation).finally(releaseSchedule);
289 1 : }, [dispatchComputation]);
290 :
291 1 : return {
292 1 : crossFilters,
293 1 : histogramFilter,
294 1 : mostFrequentValueFilter,
295 1 : requestFilteredColumnAggregation,
296 1 : };
297 1 : }
298 :
299 : /// Selection row counts per filter table and data search table.
300 : /// Weakly keyed by both Arrow tables, so a count never outlives the tables it was computed from.
301 1 : const selectionRowCounts = new WeakMap<object, WeakMap<object, number>>();
302 :
303 1 : function countSelectionRows(filterTable: FilterTable | null, dataSearchTable: DataSearchTable | null): number {
304 1 : if (filterTable == null) return dataSearchTable?.dataTable.numRows ?? 0;
305 1 : if (dataSearchTable == null) return filterTable.dataTable.numRows;
306 1 : let countsBySearch = selectionRowCounts.get(filterTable.dataTable);
307 1 : const cached = countsBySearch?.get(dataSearchTable.dataTable);
308 1 : if (cached !== undefined) return cached;
309 1 : const searchRows = dataSearchTable.dataTable.getChildAt(0);
310 1 : const filterRows = filterTable.dataTable.getChildAt(0);
311 1 : if (searchRows == null || filterRows == null) return 0;
312 1 : const matches = new Set<number>();
313 1 : forEachIntegerValue(searchRows, rowId => matches.add(rowId));
314 1 : let count = 0;
315 1 : forEachIntegerValue(filterRows, rowId => {
316 1 : if (matches.has(rowId)) ++count;
317 1 : });
318 1 : if (countsBySearch == null) {
319 1 : countsBySearch = new WeakMap();
320 1 : selectionRowCounts.set(filterTable.dataTable, countsBySearch);
321 1 : }
322 1 : countsBySearch.set(dataSearchTable.dataTable, count);
323 1 : return count;
324 1 : }
325 :
326 1 : const EMPTY_CROSS_FILTERS = new CrossFilters();
327 1 : const EMPTY_COLUMN_GROUPS: TableComputationState['columnGroups'] = [];
|
