Line data Source code
1 : import * as arrow from 'apache-arrow';
2 : import type { Spec as VegaSpec } from 'vega';
3 : import type { TopLevelSpec } from 'vega-lite';
4 :
5 1 : const SOURCE_DATASET_BASE = '__dashql_source';
6 1 : const MASK_DATASET_BASE = '__dashql_crossfilter_ids';
7 1 : const MASK_ACTIVE_BASE = '__dashql_crossfilter_active';
8 1 : const MASK_SELECTED_FIELD_BASE = '__dashql_crossfilter_selected';
9 1 : const STABLE_DOMAIN_SIGNAL_BASE = '__dashql_stable_domain';
10 :
11 1 : export const MASK_ROW_ID_FIELD = '__dashql_row_id';
12 :
13 : export interface VegaLiteCrossFilterBinding {
14 : spec: TopLevelSpec;
15 : sourceDatasetName: string;
16 : maskDatasetName: string | null;
17 : maskActiveSignalName: string | null;
18 : maskSelectedFieldName: string | null;
19 : }
20 :
21 : export interface VegaCrossFilterView {
22 : data(name: string, values: object[]): VegaCrossFilterView;
23 : signal(name: string, value: unknown): VegaCrossFilterView;
24 : scale(name: string): { domain(): unknown[] };
25 : runAsync(): Promise<unknown>;
26 : }
27 :
28 : export interface VegaStableScaleDomain {
29 : scaleName: string;
30 : signalName: string;
31 : }
32 :
33 : export interface VegaStableScaleBinding {
34 : spec: VegaSpec;
35 : domains: VegaStableScaleDomain[];
36 : }
37 :
38 : export interface VegaCrossFilterMask {
39 : active: boolean;
40 : rows: VegaCrossFilterRow[];
41 : }
42 :
43 : export interface VegaCrossFilterRow {
44 : [MASK_ROW_ID_FIELD]: number;
45 : [key: string]: number | boolean;
46 : }
47 :
48 1 : function allocateName(base: string, usedNames: Set<string>): string {
49 1 : let name = base;
50 1 : let suffix = 2;
51 1 : while (usedNames.has(name)) {
52 1 : name = `${base}_${suffix++}`;
53 : }
54 1 : usedNames.add(name);
55 1 : return name;
56 1 : }
57 :
58 1 : function collectRuntimeNames(value: unknown, names: Set<string>): void {
59 1 : if (Array.isArray(value)) {
60 1 : for (const entry of value) {
61 1 : collectRuntimeNames(entry, names);
62 : }
63 1 : return;
64 1 : }
65 1 : if (value == null || typeof value !== 'object') {
66 1 : return;
67 : }
68 1 : const object = value as Record<string, unknown>;
69 1 : if (typeof object.name === 'string') {
70 1 : names.add(object.name);
71 : }
72 1 : for (const entry of Object.values(object)) {
73 1 : collectRuntimeNames(entry, names);
74 : }
75 1 : }
76 :
77 1 : function hasDataReference(value: unknown): boolean {
78 1 : if (Array.isArray(value)) {
79 1 : return value.some(hasDataReference);
80 : }
81 1 : if (value == null || typeof value !== 'object') {
82 1 : return false;
83 : }
84 1 : const object = value as Record<string, unknown>;
85 1 : if (typeof object.data === 'string') {
86 1 : return true;
87 : }
88 0 : return Object.values(object).some(hasDataReference);
89 1 : }
90 :
91 : /// Override compiled, data-driven Vega scale domains through initially-null signals.
92 : /// The updater fills these signals after evaluating the complete source once, so later
93 : /// cross-filter pulses can recompute marks without changing axes or categorical mappings.
94 1 : export function injectVegaStableScaleDomains(spec: VegaSpec): VegaStableScaleBinding {
95 1 : const input = spec as VegaSpec & {
96 : scales?: Array<Record<string, unknown>>;
97 : signals?: Array<Record<string, unknown>>;
98 : };
99 1 : const usedNames = new Set<string>();
100 1 : collectRuntimeNames(input, usedNames);
101 1 : const domains: VegaStableScaleDomain[] = [];
102 1 : const signals = [...(input.signals ?? [])];
103 1 : const scales = (input.scales ?? []).map(scale => {
104 1 : const scaleName = scale.name;
105 1 : if (typeof scaleName !== 'string' || scale.domainRaw != null || !hasDataReference(scale.domain)) {
106 1 : return scale;
107 : }
108 1 : const signalName = allocateName(STABLE_DOMAIN_SIGNAL_BASE, usedNames);
109 1 : domains.push({ scaleName, signalName });
110 1 : signals.push({ name: signalName, value: null });
111 1 : return {
112 1 : ...scale,
113 1 : domainRaw: { signal: signalName },
114 1 : };
115 1 : });
116 :
117 1 : if (domains.length === 0) {
118 0 : return { spec, domains };
119 : }
120 1 : return {
121 1 : spec: {
122 1 : ...input,
123 1 : scales,
124 1 : signals,
125 1 : },
126 1 : domains,
127 1 : };
128 1 : }
129 :
130 : /// Add the renderer-owned row-id relation and membership lookup/filter to a runtime copy
131 : /// of a Vega-Lite spec. Both are prepended so authored transforms only see selected rows.
132 1 : export function injectVegaLiteCrossFilter(
133 1 : spec: TopLevelSpec,
134 1 : rowNumberFieldName: string | null,
135 1 : sourceFieldNames: Iterable<string> = [],
136 1 : ): VegaLiteCrossFilterBinding {
137 1 : const input = spec as TopLevelSpec & {
138 : datasets?: Record<string, unknown>;
139 : params?: Array<{ name?: string }>;
140 : transform?: unknown[];
141 : };
142 1 : const usedNames = new Set<string>(Object.keys(input.datasets ?? {}));
143 1 : collectRuntimeNames(input, usedNames);
144 1 : for (const fieldName of sourceFieldNames) {
145 1 : usedNames.add(fieldName);
146 : }
147 1 : for (const param of input.params ?? []) {
148 1 : if (param.name != null) {
149 1 : usedNames.add(param.name);
150 : }
151 : }
152 :
153 1 : const sourceDatasetName = allocateName(SOURCE_DATASET_BASE, usedNames);
154 1 : const maskDatasetName = rowNumberFieldName != null ? allocateName(MASK_DATASET_BASE, usedNames) : null;
155 1 : const maskActiveSignalName = rowNumberFieldName != null ? allocateName(MASK_ACTIVE_BASE, usedNames) : null;
156 1 : const maskSelectedFieldName = rowNumberFieldName != null ? allocateName(MASK_SELECTED_FIELD_BASE, usedNames) : null;
157 1 : const datasets = { ...(input.datasets ?? {}) };
158 1 : const params = [...(input.params ?? [])];
159 1 : const transforms = [...(input.transform ?? [])];
160 :
161 1 : if (rowNumberFieldName != null && maskDatasetName != null && maskActiveSignalName != null && maskSelectedFieldName != null) {
162 1 : datasets[maskDatasetName] = [];
163 1 : params.push({ name: maskActiveSignalName, value: false });
164 1 : transforms.unshift(
165 1 : {
166 1 : lookup: rowNumberFieldName,
167 1 : from: {
168 1 : data: { name: maskDatasetName },
169 1 : key: MASK_ROW_ID_FIELD,
170 1 : fields: [maskSelectedFieldName],
171 1 : },
172 1 : as: [maskSelectedFieldName],
173 1 : default: false,
174 1 : },
175 1 : {
176 1 : filter: `!${maskActiveSignalName} || datum[${JSON.stringify(maskSelectedFieldName)}] === true`,
177 1 : },
178 1 : );
179 1 : }
180 : const augmented = {
181 1 : ...input,
182 1 : data: { name: sourceDatasetName },
183 1 : datasets,
184 1 : params,
185 1 : transform: transforms,
186 : } as unknown as TopLevelSpec;
187 :
188 1 : return {
189 1 : spec: augmented,
190 1 : sourceDatasetName,
191 1 : maskDatasetName,
192 1 : maskActiveSignalName,
193 1 : maskSelectedFieldName,
194 1 : };
195 1 : }
196 :
197 : /// Convert the one-column DashQL filter table to the small relation consumed by the lookup.
198 1 : export function filterTableToVegaCrossFilterRows(table: arrow.Table, selectedFieldName: string): VegaCrossFilterRow[] {
199 1 : if (table.numCols !== 1) {
200 1 : return [];
201 : }
202 1 : const column = table.getChildAt(0);
203 1 : if (column == null) {
204 0 : return [];
205 : }
206 1 : const rows: VegaCrossFilterRow[] = [];
207 1 : for (let i = 0; i < column.length; ++i) {
208 1 : const value = column.get(i);
209 1 : if (value == null) {
210 0 : continue;
211 : }
212 1 : rows.push({
213 1 : [MASK_ROW_ID_FIELD]: Number(value),
214 1 : [selectedFieldName]: true,
215 1 : });
216 1 : }
217 1 : return rows;
218 1 : }
219 :
220 : /// Serialize Vega runs and collapse rapid brush results to the newest pending mask.
221 1 : export class VegaCrossFilterUpdater {
222 1 : private view: VegaCrossFilterView | null;
223 1 : private sourceRows: object[] | null;
224 1 : private readonly sourceDatasetName: string;
225 1 : private readonly maskDatasetName: string | null;
226 1 : private readonly maskActiveSignalName: string | null;
227 1 : private readonly stableScaleDomains: readonly VegaStableScaleDomain[];
228 1 : private pending: VegaCrossFilterMask | null = null;
229 1 : private running = false;
230 1 : private readonly onError: (error: unknown) => void;
231 :
232 1 : constructor(
233 1 : view: VegaCrossFilterView,
234 1 : sourceDatasetName: string,
235 1 : sourceRows: object[],
236 1 : maskDatasetName: string | null,
237 1 : maskActiveSignalName: string | null,
238 1 : onError: (error: unknown) => void = () => { },
239 1 : stableScaleDomains: readonly VegaStableScaleDomain[] = [],
240 1 : ) {
241 1 : this.view = view;
242 1 : this.sourceDatasetName = sourceDatasetName;
243 1 : this.sourceRows = sourceRows;
244 1 : this.maskDatasetName = maskDatasetName;
245 1 : this.maskActiveSignalName = maskActiveSignalName;
246 1 : this.onError = onError;
247 1 : this.stableScaleDomains = stableScaleDomains;
248 1 : }
249 :
250 1 : update(mask: VegaCrossFilterMask): void {
251 1 : if (this.view == null) {
252 0 : return;
253 : }
254 1 : this.pending = mask;
255 1 : if (!this.running) {
256 1 : void this.flush();
257 : }
258 1 : }
259 :
260 1 : dispose(): void {
261 1 : this.view = null;
262 1 : this.sourceRows = null;
263 1 : this.pending = null;
264 1 : }
265 :
266 1 : private async flush(): Promise<void> {
267 1 : this.running = true;
268 1 : try {
269 1 : while (this.view != null && this.pending != null) {
270 1 : const view = this.view;
271 : // Did the run of the scale domains show the source without a mask?
272 1 : let unmaskedRun = false;
273 1 : if (this.sourceRows != null) {
274 1 : view.data(this.sourceDatasetName, this.sourceRows);
275 1 : this.sourceRows = null;
276 1 : if (this.stableScaleDomains.length > 0) {
277 1 : if (this.maskDatasetName != null && this.maskActiveSignalName != null) {
278 1 : view
279 1 : .data(this.maskDatasetName, [])
280 1 : .signal(this.maskActiveSignalName, false);
281 : }
282 1 : await view.runAsync();
283 1 : if (this.view !== view) {
284 0 : continue;
285 : }
286 1 : for (const domain of this.stableScaleDomains) {
287 1 : view.signal(domain.signalName, [...view.scale(domain.scaleName).domain()]);
288 : }
289 1 : unmaskedRun = true;
290 1 : }
291 1 : }
292 1 : const mask = this.pending;
293 1 : if (mask == null) {
294 0 : continue;
295 : }
296 1 : this.pending = null;
297 : // A chart without a filter already shows what a second run would show, the domain signals
298 : // take effect with the next run. A second run doubles the work of every chart that mounts.
299 1 : if (unmaskedRun && !mask.active && mask.rows.length === 0) {
300 1 : continue;
301 : }
302 1 : if (this.maskDatasetName != null && this.maskActiveSignalName != null) {
303 1 : view
304 1 : .data(this.maskDatasetName, mask.rows)
305 1 : .signal(this.maskActiveSignalName, mask.active);
306 : }
307 1 : await view.runAsync();
308 1 : }
309 1 : } catch (error: unknown) {
310 0 : this.onError(error);
311 1 : } finally {
312 1 : this.running = false;
313 1 : if (this.view != null && this.pending != null) {
314 0 : void this.flush();
315 : }
316 1 : }
317 1 : }
318 1 : }
|
