Line data Source code
1 : import { ArrowTableFormatter } from '../arrow_formatter.js';
2 : import { DataFrame, generateTableName } from '../data_frame.js';
3 : import { ColumnAggregationVariant, ColumnAggregationTask, ORDINAL_COLUMN, STRING_COLUMN, LIST_COLUMN, SKIPPED_COLUMN, ROWNUMBER_COLUMN, getGridColumnTypeName, WithFilter, WithFilterEpoch } from '../computation_types.js';
4 : import { LoggableException, LoggerLike, stringifyError } from '../../platform/logger/logger.js';
5 : import { OrderByConstraint, SQLFrame } from '../sql/sqlframe_builder.js';
6 : import { ComputeQueryExecution, executeDirectly } from '../computation_tasks.js';
7 : import { getToNumericFn, isIntegerType } from '../arrow_columns.js';
8 : import { analyzeOrdinalColumn, analyzeStringColumn } from '../column_analysis.js';
9 :
10 1 : const LOG_CTX = "compute";
11 :
12 : /// The order of the bins of an ordinal column
13 1 : const BIN_ORDER: OrderByConstraint[] = [{ field: "bin", ascending: true }];
14 : /// The order of the frequent values of a string column, most frequent first
15 1 : const FREQUENT_VALUE_ORDER: OrderByConstraint[] = [{ field: "count", ascending: false }];
16 : /// The number of frequent values of a string column
17 1 : const FREQUENT_VALUE_COUNT = 32;
18 :
19 1 : export async function computeColumnAggregates(task: ColumnAggregationTask, logger: LoggerLike, executeQuery: ComputeQueryExecution = executeDirectly): Promise<ColumnAggregationVariant> {
20 1 : if (task.columnEntry.type == SKIPPED_COLUMN || task.columnEntry.type == ROWNUMBER_COLUMN || task.columnEntry.type == LIST_COLUMN) {
21 0 : throw new LoggableException(`Column of type cannot be aggregated`, {
22 0 : type: getGridColumnTypeName(task.columnEntry),
23 0 : }, LOG_CTX);
24 : }
25 :
26 1 : const [sql, orderBy] = buildColumnAggregationSQL(task);
27 :
28 1 : try {
29 1 : const transformStart = performance.now();
30 1 : const tableName = generateTableName("__col_agg");
31 1 : const aggregateDataFrame = await executeQuery(sql, clientContext => DataFrame.fromSQL(task.inputDataFrame.database, sql, tableName, undefined, clientContext));
32 1 : const transformEnd = performance.now();
33 :
34 1 : const aggregateTable = await aggregateDataFrame.readTable({ orderBy });
35 1 : const aggregateTableFormatter = new ArrowTableFormatter(aggregateTable.schema, aggregateTable.batches, logger);
36 1 : const readEnd = performance.now();
37 :
38 1 : let columnAggregate: ColumnAggregationVariant;
39 1 : switch (task.columnEntry.type) {
40 1 : case ORDINAL_COLUMN: {
41 1 : const analysis = analyzeOrdinalColumn(task.tableAggregate, task.columnEntry.value, aggregateTable, aggregateTableFormatter);
42 1 : columnAggregate = {
43 1 : type: ORDINAL_COLUMN,
44 1 : value: {
45 1 : columnEntry: task.columnEntry.value,
46 1 : binnedDataFrame: aggregateDataFrame,
47 1 : binnedValues: aggregateTable,
48 1 : binnedValuesFormatter: aggregateTableFormatter,
49 1 : columnAnalysis: analysis,
50 1 : }
51 1 : };
52 1 : break;
53 1 : }
54 1 : case STRING_COLUMN: {
55 1 : const analysis = analyzeStringColumn(task.tableAggregate, task.columnEntry.value, aggregateTable, aggregateTableFormatter);
56 1 : columnAggregate = {
57 1 : type: STRING_COLUMN,
58 1 : value: {
59 1 : columnEntry: task.columnEntry.value,
60 1 : frequentValuesDataFrame: aggregateDataFrame,
61 1 : frequentValuesTable: aggregateTable,
62 1 : frequentValuesFormatter: aggregateTableFormatter,
63 1 : analysis,
64 1 : }
65 1 : };
66 1 : break;
67 1 : }
68 1 : }
69 1 : const analysisEnd = performance.now();
70 1 : logger.info("Aggregated table column", {
71 1 : "version": task.tableVersion.toString(),
72 1 : "table": task.tableId.toString(),
73 1 : "columnIndex": task.columnId.toString(),
74 1 : "columnName": task.columnEntry.value.inputFieldName,
75 1 : "groupType": getGridColumnTypeName(task.columnEntry),
76 1 : "duration": Math.floor(analysisEnd - transformStart).toString(),
77 1 : "queryDuration": Math.floor(transformEnd - transformStart).toString(),
78 1 : "readDuration": Math.floor(readEnd - transformEnd).toString(),
79 1 : "analysisDuration": Math.floor(analysisEnd - readEnd).toString(),
80 1 : "resultRows": aggregateTable.numRows.toString(),
81 1 : "sql": sql,
82 1 : }, LOG_CTX);
83 1 : return columnAggregate;
84 :
85 1 : } catch (error: any) {
86 0 : if (error instanceof LoggableException) {
87 0 : throw error;
88 0 : } else {
89 0 : const exception = new LoggableException("Computing column aggregate failed", {
90 0 : "table": task.tableId.toString(),
91 0 : "error": stringifyError(error),
92 0 : }, LOG_CTX);
93 0 : logger.warn(exception.message, exception.keyValues, exception.target);
94 0 : throw exception;
95 0 : }
96 0 : }
97 1 : }
98 :
99 : /// The SQL of the aggregates of a column, and the order of their rows
100 1 : function buildColumnAggregationSQL(task: ColumnAggregationTask, filtered: WithFilter<ColumnAggregationTask> | null = null): [string, OrderByConstraint[]] {
101 1 : if (task.columnEntry.type == SKIPPED_COLUMN || task.columnEntry.type == ROWNUMBER_COLUMN || task.columnEntry.value.statsFields == null) {
102 0 : throw new Error("Column summary requires precomputed table summary");
103 : }
104 :
105 1 : let frame = SQLFrame.from(task.inputDataFrame.tableName);
106 :
107 1 : if (filtered != null) {
108 1 : if (filtered.filterTable != null) {
109 1 : frame = frame.semiJoinFilter(
110 1 : filtered.filterTable.inputRowNumberColumnName,
111 1 : filtered.filterTable.dataFrame.tableName,
112 1 : filtered.filterTable.dataTable.schema.fields[0].name,
113 1 : );
114 : }
115 1 : if (filtered.dataSearchTable != null) {
116 1 : frame = frame.semiJoinFilter(
117 1 : filtered.dataSearchTable.inputRowNumberColumnName,
118 1 : filtered.dataSearchTable.dataFrame.tableName,
119 1 : filtered.dataSearchTable.dataTable.schema.fields[0].name,
120 1 : );
121 : }
122 1 : }
123 :
124 1 : const targetFieldName = task.columnEntry.value.inputFieldName;
125 :
126 1 : switch (task.columnEntry.type) {
127 1 : case ORDINAL_COLUMN: {
128 1 : const minField = task.columnEntry.value.statsFields.minAggregateFieldName!;
129 1 : const maxField = task.columnEntry.value.statsFields.maxAggregateFieldName!;
130 1 : frame = frame.groupBy({
131 1 : keys: [{
132 1 : fieldName: targetFieldName,
133 1 : outputAlias: "bin",
134 1 : binning: {
135 1 : preBinnedFieldName: task.columnEntry.value.binFieldName ?? undefined,
136 1 : statsTable: task.tableAggregate.dataFrame.tableName,
137 1 : statsMinField: minField,
138 1 : statsMaxField: maxField,
139 1 : binCount: task.columnEntry.value.binCount,
140 1 : minBinWidth: isIntegerType(task.columnEntry.value.inputFieldType.typeId) ? 1 : undefined,
141 1 : outputBinWidthAlias: "binWidth",
142 1 : outputBinLbAlias: "binLowerBound",
143 1 : outputBinUbAlias: "binUpperBound",
144 1 : includeNullBin: true,
145 1 : toNumericFn: getToNumericFn(task.columnEntry.value.inputFieldType.typeId),
146 1 : }
147 1 : }],
148 1 : aggregates: [{
149 1 : func: "count_star",
150 1 : outputAlias: "count",
151 1 : }]
152 1 : });
153 1 : return [frame.toSQL(), BIN_ORDER];
154 1 : }
155 1 : case STRING_COLUMN: {
156 1 : if (task.columnEntry.value.valueIdFieldName == null) {
157 0 : throw new Error("Cannot aggregate string column without precomputed value id");
158 : }
159 1 : frame = frame.groupBy({
160 1 : keys: [
161 1 : { fieldName: targetFieldName, outputAlias: "key" },
162 1 : { fieldName: task.columnEntry.value.valueIdFieldName, outputAlias: "keyId" },
163 1 : ],
164 1 : aggregates: [{
165 1 : func: "count_star",
166 1 : outputAlias: "count",
167 1 : }]
168 1 : }).orderBy(FREQUENT_VALUE_ORDER, FREQUENT_VALUE_COUNT);
169 1 : return [frame.toSQL(), FREQUENT_VALUE_ORDER];
170 : }
171 1 : }
172 :
173 0 : return [frame.toSQL(), []];
174 1 : }
175 :
176 1 : export async function computeFilteredColumnAggregates(task: WithFilter<ColumnAggregationTask>, logger: LoggerLike, executeQuery: ComputeQueryExecution = executeDirectly): Promise<WithFilterEpoch<ColumnAggregationVariant> | null> {
177 1 : if (task.columnEntry.type == SKIPPED_COLUMN || task.columnEntry.type == ROWNUMBER_COLUMN || task.columnEntry.type == LIST_COLUMN) {
178 1 : throw new LoggableException(`Column of type cannot be aggregated`, {
179 1 : type: getGridColumnTypeName(task.columnEntry),
180 1 : }, LOG_CTX);
181 : }
182 :
183 1 : const [sql, orderBy] = buildColumnAggregationSQL(task, task);
184 :
185 1 : try {
186 1 : logger.info("Aggregating filtered table column", {
187 1 : "version": task.tableVersion.toString(),
188 1 : "table": task.tableId.toString(),
189 1 : "columnIndex": task.columnId.toString(),
190 1 : "columnName": task.columnEntry.value.inputFieldName,
191 1 : "groupType": getGridColumnTypeName(task.columnEntry),
192 1 : }, LOG_CTX);
193 1 : const transformStart = performance.now();
194 1 : const tableName = generateTableName("__filt_col_agg");
195 1 : const aggregateDataFrame = await executeQuery(sql, clientContext => DataFrame.fromSQL(task.inputDataFrame.database, sql, tableName, undefined, clientContext));
196 1 : const transformEnd = performance.now();
197 1 : logger.info("Aggregated filtered table column", {
198 1 : "version": task.tableVersion.toString(),
199 1 : "table": task.tableId.toString(),
200 1 : "columnIndex": task.columnId.toString(),
201 1 : "columnName": task.columnEntry.value.inputFieldName,
202 1 : "groupType": getGridColumnTypeName(task.columnEntry),
203 1 : "duration": Math.floor(transformEnd - transformStart).toString(),
204 1 : "sql": sql
205 1 : }, LOG_CTX);
206 :
207 1 : const aggregateTable = await aggregateDataFrame.readTable({ orderBy });
208 1 : const aggregateTableFormatter = new ArrowTableFormatter(aggregateTable.schema, aggregateTable.batches, logger);
209 :
210 1 : let columnAggregate: WithFilterEpoch<ColumnAggregationVariant>;
211 1 : switch (task.columnEntry.type) {
212 1 : case ORDINAL_COLUMN: {
213 1 : const analysis = analyzeOrdinalColumn(task.tableAggregate, task.columnEntry.value, aggregateTable, aggregateTableFormatter, task.selectionRowCount);
214 1 : columnAggregate = {
215 1 : type: ORDINAL_COLUMN,
216 1 : value: {
217 1 : columnEntry: task.columnEntry.value,
218 1 : binnedDataFrame: aggregateDataFrame,
219 1 : binnedValues: aggregateTable,
220 1 : binnedValuesFormatter: aggregateTableFormatter,
221 1 : columnAnalysis: analysis,
222 1 : },
223 1 : filterVersion: task.tableVersion.clone(),
224 1 : dataSearchRequestId: task.dataSearchTable?.requestId ?? null,
225 1 : };
226 1 : break;
227 1 : }
228 1 : case STRING_COLUMN: {
229 1 : const analysis = analyzeStringColumn(task.tableAggregate, task.columnEntry.value, aggregateTable, aggregateTableFormatter, task.selectionRowCount);
230 1 : columnAggregate = {
231 1 : type: STRING_COLUMN,
232 1 : value: {
233 1 : columnEntry: task.columnEntry.value,
234 1 : frequentValuesDataFrame: aggregateDataFrame,
235 1 : frequentValuesTable: aggregateTable,
236 1 : frequentValuesFormatter: aggregateTableFormatter,
237 1 : analysis,
238 1 : },
239 1 : filterVersion: task.tableVersion.clone(),
240 1 : dataSearchRequestId: task.dataSearchTable?.requestId ?? null,
241 1 : };
242 1 : break;
243 1 : }
244 1 : }
245 1 : return columnAggregate;
246 :
247 1 : } catch (error: any) {
248 1 : if (error instanceof LoggableException) {
249 0 : throw error;
250 1 : } else {
251 1 : const exception = new LoggableException("Failed to compute filtered column aggregate", {
252 1 : "table": task.tableId.toString(),
253 1 : "columnIndex": task.columnId.toString(),
254 1 : "columnName": task.columnEntry.value.inputFieldName,
255 1 : "error": stringifyError(error),
256 1 : }, LOG_CTX);
257 1 : logger.warn(exception.message, exception.keyValues, exception.target);
258 1 : throw exception;
259 1 : }
260 1 : }
261 1 : }
|
