Line data Source code
1 : import * as arrow from 'apache-arrow';
2 :
3 : import { DataFrame, generateTableName } from './data_frame.js';
4 : import { COLUMN_AGGREGATION_TASK, SYSTEM_COLUMN_COMPUTATION_TASK, TABLE_AGGREGATION_TASK } from './computation_types.js';
5 : import { COMPUTATION_FROM_QUERY_RESULT, ComputationAction, CREATED_DATA_FRAME } from './computation_state.js';
6 : import { ColumnAggregationTask, TableAggregationTask, ORDINAL_COLUMN, STRING_COLUMN, LIST_COLUMN, ColumnGroup, SKIPPED_COLUMN, SystemColumnComputationTask, ROWNUMBER_COLUMN, ComputationStateVersion } from './computation_types.js';
7 : import { Dispatch } from '../utils/variant.js';
8 : import { LoggerLike } from '../platform/logger/logger.js';
9 : import type { EmbeddedClientContext, EmbeddedComputeDatabase } from '../platform/database/embedded_database.js';
10 : import { scheduleTask } from './computation_tasks.js';
11 : import { createUniqueColumnName } from './arrow_columns.js';
12 :
13 1 : const LOG_CTX = "compute";
14 :
15 : /// The number of bins of an ordinal column
16 1 : const BIN_COUNT = 16;
17 :
18 : /// Options of the analysis of a table
19 : export interface AnalyzeTableOptions {
20 : /// The client context of the queries of a table without a connection, e.g. of the connection overview
21 : clientContext?: EmbeddedClientContext | null;
22 : /// Compute the aggregates of the columns, e.g. false for a table that shows no column plots
23 : columnAggregates?: boolean;
24 : }
25 :
26 : /// Analyze a table.
27 : ///
28 : /// This function computes multiple summaries for displaying the data in the table grid component.
29 : ///
30 : /// It consists of multiple steps:
31 : /// 1) We first global table aggregates.
32 : /// For each of the columns we compute minimum, maximum, counts and distinct counts depending on the column type.
33 : /// 2) We then compute auxiliary system columns storing the input row number and (fractional) bin values in the domain.
34 : /// 3) We then compute summary tables per column based on these bin values.
35 : /// (For example an aggregated histogram table for ordinal columns)
36 : ///
37 : /// On Cross-Filters:
38 : /// One option for fast cross-filters is to pre-compute summed area tables from Dominik's Falcon paper:
39 : /// https://osf.io/preprints/osf/szpqm_v1
40 : /// This paper proposes to precompute a summation table to assist brushing.
41 : /// The idea is the following:
42 : /// A) Aggregate counts for the cube of all cross-filter dimensions (N table columns -> 2^N fields).
43 : /// (A, B, C) -> grouping sets ((A, B, C), (A, B), (A, C), (B, C), (A), (B), (C), ())
44 : /// B) Compute comulative sums for the aggregates.
45 : /// When brushing, we want to compute the count between two boundaries, for which we need cumulative sums.
46 : /// C) Cross-filtering then becomes a O(1) lookup in this cube.
47 : ///
48 : /// Problem:
49 : /// This is simple when doing this for ordinal columns only, where we control the dimension count.
50 : /// For categorical columns, we have the problem that we might want to cross-filter with a *specific* distinct value.
51 : /// Precomputing for that is infeasible with higher distinct counts, so we'll have to maintain a non-precomputed path anyway?
52 : /// Also, 2^N is not exactly cheap for very wide tables where the user would never visibly see the cross-filter.
53 : ///
54 : /// Alternative:
55 : /// We do ad-hoc cross-filter computations on the pre-computed bin fields and only for what is currently visible (!).
56 : /// The upside here is that the bin count is dictated by what we can show in the UI and is thus usually small.
57 : /// Whenever a user updates a cross-filter (by brushing or selecting a distinct value), we just recompute the column summaries
58 : /// with the new set of cross-filters and update the UI.
59 1 : export async function analyzeTable(tableId: number, table: arrow.Table, dispatch: Dispatch<ComputationAction>, database: EmbeddedComputeDatabase, logger: LoggerLike, connectionId: string | null = null, options: AnalyzeTableOptions = {}): Promise<void> {
60 : // Statements such as DDL can legitimately produce an Arrow table without columns. There is
61 : // nothing to summarize, and embedded databases reject importing a zero-column table.
62 1 : if (table.numCols === 0) {
63 1 : return;
64 : }
65 :
66 1 : let gridColumnGroups = buildGridColumnGroups(table!);
67 1 : const computeAbortCtrl = new AbortController();
68 1 : dispatch({
69 1 : type: COMPUTATION_FROM_QUERY_RESULT,
70 1 : value: [tableId, table!, gridColumnGroups, computeAbortCtrl, connectionId, options.clientContext ?? null]
71 1 : });
72 :
73 1 : const [numberedTable, rowNumberFieldName] = appendRowNumbers(table);
74 1 : const inputTableName = generateTableName("__input");
75 1 : let dataFrame = await DataFrame.fromArrowTable(database, numberedTable, inputTableName, options.clientContext ?? null);
76 1 : dispatch({
77 1 : type: CREATED_DATA_FRAME,
78 1 : value: [tableId, dataFrame]
79 1 : });
80 :
81 : // Compute table aggregates
82 1 : const tableAggregationTask: TableAggregationTask = {
83 1 : tableId,
84 1 : tableVersion: new ComputationStateVersion(0, 0),
85 1 : columnEntries: gridColumnGroups,
86 1 : inputDataFrame: dataFrame
87 1 : };
88 1 : const [tableAggregate, initialColumnGroups] = await scheduleTask(TABLE_AGGREGATION_TASK, tableAggregationTask, dispatch);
89 1 : gridColumnGroups = initialColumnGroups;
90 :
91 : // Precompute the system columns
92 1 : const precomputationTask: SystemColumnComputationTask = {
93 1 : tableId,
94 1 : tableVersion: new ComputationStateVersion(0, 0),
95 1 : columnEntries: gridColumnGroups,
96 1 : inputTable: table,
97 1 : inputDataFrame: dataFrame,
98 1 : rowNumberFieldName,
99 1 : tableAggregate
100 1 : };
101 1 : const [newTable, newDataFrame, updatedColumnGroups] = await scheduleTask(SYSTEM_COLUMN_COMPUTATION_TASK, precomputationTask, dispatch);
102 1 : gridColumnGroups = updatedColumnGroups;
103 :
104 : // Compute the column aggregates
105 1 : let colAggregateCount = 0;
106 1 : let colAggregateOrdinalCount = 0;
107 1 : let colAggregateStringCount = 0;
108 1 : let colAggregateSlowestDuration = 0;
109 1 : let colAggregateSlowestColumn: string | null = null;
110 1 : let colAggregatesBegin = performance.now();
111 1 : logger.info("Computing column aggregates", {
112 1 : "table": tableId.toString(),
113 1 : "rows": newTable.numRows.toString(),
114 1 : "columns": gridColumnGroups.length.toString(),
115 1 : }, LOG_CTX);
116 : // A table without column plots needs no column aggregates
117 1 : const aggregatedColumns = options.columnAggregates === false ? 0 : gridColumnGroups.length;
118 1 : for (let columnId = 0; columnId < aggregatedColumns; ++columnId) {
119 : // LIST columns (e.g. float32 embedding arrays) are not aggregated: their value
120 : // identifier / frequent-value / distinct-count computations are very expensive
121 : // (full-vector sort/hash) and buy nothing — no summary histogram is rendered for
122 : // them. We keep the LIST_COLUMN group but skip all of its embedded-database
123 : // aggregation work.
124 1 : const columnEntry = gridColumnGroups[columnId];
125 1 : if (columnEntry.type == SKIPPED_COLUMN || columnEntry.type == ROWNUMBER_COLUMN || columnEntry.type == LIST_COLUMN) {
126 1 : continue;
127 : }
128 1 : const columnAggregationTask: ColumnAggregationTask = {
129 1 : tableId,
130 1 : tableVersion: new ComputationStateVersion(0, 0),
131 1 : columnId,
132 1 : tableAggregate,
133 1 : columnEntry,
134 1 : inputDataFrame: newDataFrame,
135 1 : };
136 : // Measured around the dispatch, so this includes the scheduler round trip.
137 1 : const columnBegin = performance.now();
138 1 : await scheduleTask(COLUMN_AGGREGATION_TASK, columnAggregationTask, dispatch);
139 1 : const columnDuration = performance.now() - columnBegin;
140 1 : if (columnDuration > colAggregateSlowestDuration) {
141 1 : colAggregateSlowestDuration = columnDuration;
142 1 : colAggregateSlowestColumn = columnEntry.value.inputFieldName;
143 1 : }
144 1 : ++colAggregateCount;
145 1 : if (columnEntry.type == ORDINAL_COLUMN) {
146 1 : ++colAggregateOrdinalCount;
147 : } else {
148 1 : ++colAggregateStringCount;
149 : }
150 1 : }
151 1 : let colAggregatesEnd = performance.now();
152 :
153 1 : if (colAggregateCount > 0) {
154 1 : const duration = colAggregatesEnd - colAggregatesBegin;
155 1 : logger.info("Computed column aggregates", {
156 1 : "table": tableId.toString(),
157 1 : "rows": newTable.numRows.toString(),
158 1 : "aggregates": colAggregateCount.toString(),
159 1 : "ordinal": colAggregateOrdinalCount.toString(),
160 1 : "string": colAggregateStringCount.toString(),
161 1 : "duration": Math.floor(duration).toString(),
162 1 : "avgDuration": Math.floor(duration / colAggregateCount).toString(),
163 1 : "maxDuration": Math.floor(colAggregateSlowestDuration).toString(),
164 1 : "slowestColumn": colAggregateSlowestColumn ?? "",
165 1 : }, LOG_CTX);
166 1 : }
167 1 : }
168 :
169 : /// A table with the positions of its rows as an appended row-number column, from 1, and the name of the column.
170 : /// The row numbers are appended before the import into the embedded database because Hyper numbers the rows of
171 : /// row_number() OVER () in no particular order.
172 1 : export function appendRowNumbers(table: arrow.Table): [arrow.Table, string] {
173 1 : const fieldName = createUniqueColumnName("_rownum", new Set(table.schema.fields.map(field => field.name)));
174 1 : const rowNumbers = BigInt64Array.from({ length: table.numRows }, (_, i) => BigInt(i + 1));
175 1 : return [table.assign(new arrow.Table({ [fieldName]: arrow.makeVector(rowNumbers) })), fieldName];
176 1 : }
177 :
178 1 : export function buildGridColumnGroups(table: arrow.Table): ColumnGroup[] {
179 1 : const columnGroups: ColumnGroup[] = [];
180 1 : for (let i = 0; i < table.schema.fields.length; ++i) {
181 1 : const field = table.schema.fields[i];
182 1 : switch (field.typeId) {
183 1 : case arrow.Type.Int:
184 1 : case arrow.Type.Int8:
185 1 : case arrow.Type.Int16:
186 1 : case arrow.Type.Int32:
187 1 : case arrow.Type.Int64:
188 1 : case arrow.Type.Uint8:
189 1 : case arrow.Type.Uint16:
190 1 : case arrow.Type.Uint32:
191 1 : case arrow.Type.Uint64:
192 1 : case arrow.Type.Float:
193 1 : case arrow.Type.Float16:
194 1 : case arrow.Type.Float32:
195 1 : case arrow.Type.Float64:
196 1 : case arrow.Type.Bool:
197 1 : case arrow.Type.Decimal:
198 1 : case arrow.Type.Date:
199 1 : case arrow.Type.DateDay:
200 1 : case arrow.Type.DateMillisecond:
201 1 : case arrow.Type.Time:
202 1 : case arrow.Type.TimeSecond:
203 1 : case arrow.Type.TimeMillisecond:
204 1 : case arrow.Type.TimeMicrosecond:
205 1 : case arrow.Type.TimeNanosecond:
206 1 : case arrow.Type.Timestamp:
207 1 : case arrow.Type.TimestampSecond:
208 1 : case arrow.Type.TimestampMillisecond:
209 1 : case arrow.Type.TimestampMicrosecond:
210 1 : case arrow.Type.TimestampNanosecond:
211 1 : case arrow.Type.DurationSecond:
212 1 : case arrow.Type.DurationMillisecond:
213 1 : case arrow.Type.DurationMicrosecond:
214 1 : case arrow.Type.DurationNanosecond:
215 1 : columnGroups.push({
216 1 : type: ORDINAL_COLUMN,
217 1 : value: {
218 1 : inputFieldName: field.name,
219 1 : inputFieldType: field.type,
220 1 : inputFieldNullable: field.nullable,
221 1 : statsFields: null,
222 1 : binFieldName: null,
223 1 : binCount: BIN_COUNT
224 1 : }
225 1 : });
226 1 : break;
227 1 : case arrow.Type.Utf8:
228 1 : case arrow.Type.LargeUtf8:
229 1 : columnGroups.push({
230 1 : type: STRING_COLUMN,
231 1 : value: {
232 1 : inputFieldName: field.name,
233 1 : inputFieldType: field.type,
234 1 : inputFieldNullable: field.nullable,
235 1 : statsFields: null,
236 1 : valueIdFieldName: null,
237 1 : }
238 1 : });
239 1 : break;
240 1 : case arrow.Type.List:
241 1 : case arrow.Type.FixedSizeList:
242 1 : columnGroups.push({
243 1 : type: LIST_COLUMN,
244 1 : value: {
245 1 : inputFieldName: field.name,
246 1 : inputFieldType: field.type,
247 1 : inputFieldNullable: field.nullable,
248 1 : statsFields: null,
249 1 : valueIdFieldName: null,
250 1 : }
251 1 : });
252 1 : break;
253 1 : default:
254 1 : columnGroups.push({
255 1 : type: SKIPPED_COLUMN,
256 1 : value: {
257 1 : inputFieldName: field.name,
258 1 : inputFieldType: field.type,
259 1 : inputFieldNullable: field.nullable,
260 1 : }
261 1 : });
262 : break;
263 1 : }
264 1 : }
265 1 : return columnGroups;
266 1 : }
|
