Line data Source code
1 : import * as arrow from 'apache-arrow';
2 : import * as proto from "../../proto.js";
3 :
4 : import { LoggableException, Logger } from '../../platform/logger/logger.js';
5 : import { createQueryResponseStreamMetrics, QueryExecutionProgress, QueryExecutionResponseStream, QueryExecutionMetrics, QueryExecutionStatus } from "../../query/query_execution_state.js";
6 : import { yieldToBrowser } from '../../utils/yield_to_browser.js';
7 : import { TrinoApiClientInterface, TrinoApiEndpoint, TrinoQueryData, TrinoQueryInfo, TrinoQueryResult, TrinoQueryStage, TrinoQueryStatistics } from "./trino_api_client.js";
8 : import { ChannelError, RawProxyError } from '../../platform/channel_common.js';
9 : import { AsyncValue } from '../../utils/async_value.js';
10 : import { AsyncConsumer } from '../../utils/async_consumer.js';
11 : import { translateAnyRowsToArrowBatch } from './arrow_conversion.js';
12 : import type { TrinoSessionDefaults } from './trino_catalogs.js';
13 :
14 1 : const LOG_CTX = 'trino_channel';
15 : /// Fetch QueryInfo again after this long while a stage whose tasks are done has no operator summaries yet
16 1 : const QUERY_INFO_RETRY_MS = 1000;
17 : /// Wait this long for the final QueryInfo of a finished query
18 1 : const QUERY_INFO_FINAL_TIMEOUT_MS = 3000;
19 : /// Fetch the QueryInfo of a finished query this often until it is final
20 1 : const QUERY_INFO_FINAL_RETRY_MS = 200;
21 :
22 : /// Are the tasks of a stage done? A stage waits for the query in PENDING and finishes with it.
23 1 : function isStageDone(state: string | undefined): boolean {
24 1 : return state === 'PENDING' || state === 'FINISHED';
25 1 : }
26 :
27 : /// The stages of the statement stats whose tasks are done
28 1 : function countDoneStages(stage: TrinoQueryStage | null | undefined): number {
29 1 : if (stage == null) return 0;
30 1 : let count = isStageDone(stage.state) ? 1 : 0;
31 1 : for (const child of stage.subStages ?? []) count += countDoneStages(child);
32 1 : return count;
33 1 : }
34 :
35 : /// Does a QueryInfo have a stage whose tasks are done but that reports no operator summaries yet? Right after a
36 : /// stage turned PENDING, Trino may still wait for the final infos of its tasks.
37 1 : function lacksOperatorSummaries(info: TrinoQueryInfo): boolean {
38 1 : return (info.stages?.stages ?? []).some(stage =>
39 1 : isStageDone(stage.state) && (stage.stageStats?.operatorSummaries?.length ?? 0) === 0);
40 1 : }
41 :
42 : /// Is a QueryInfo final? A finished query is once Trino collected the final infos of its tasks, a failed one may
43 : /// never be.
44 1 : function isFinalQueryInfo(info: TrinoQueryInfo): boolean {
45 1 : return info.state === 'FAILED' || (info.state === 'FINISHED' && info.finalQueryInfo === true);
46 1 : }
47 :
48 : export interface TrinoQueryExecutionProgress extends QueryExecutionProgress { }
49 :
50 1 : export class TrinoQueryResultStream implements QueryExecutionResponseStream {
51 : /// The logger
52 1 : logger: Logger;
53 : /// The api client
54 1 : apiClient: TrinoApiClientInterface;
55 : /// The current execution status
56 1 : currentStatus: QueryExecutionStatus;
57 : /// The response metadata
58 1 : responseMetadata: Map<string, string>;
59 : /// The schema
60 1 : resultSchema: AsyncValue<arrow.Schema, Error>;
61 : /// The current result
62 1 : latestQueryResult: TrinoQueryResult | null;
63 : /// The latest statistics
64 1 : latestQueryStats: TrinoQueryStatistics | null;
65 : /// The latest query state
66 1 : latestQueryState: string | null;
67 : /// When the latest result arrived, in milliseconds since epoch
68 1 : latestQueryReceivedAt: number;
69 : /// Was the QueryInfo with the plans of the stages requested?
70 1 : queryInfoRequested: boolean = false;
71 : /// The request of the QueryInfo with the plans
72 1 : queryInfoPlan: Promise<void> | null = null;
73 : /// The pending request of a QueryInfo for the operator summaries
74 1 : queryInfoStats: Promise<TrinoQueryInfo | null> | null = null;
75 : /// Fetch QueryInfo again once the pending request is done?
76 1 : queryInfoStale: boolean = false;
77 : /// The stages with done tasks that the requested QueryInfos cover
78 1 : queryInfoDoneStages: number = 0;
79 : /// The timer of the next QueryInfo while a stage with done tasks has no operator summaries
80 1 : queryInfoRetry: ReturnType<typeof setTimeout> | null = null;
81 : /// Is the profile shown?
82 1 : profileDetail: boolean = false;
83 : /// The consumer of the progress while the stream produces
84 1 : progressConsumer: AsyncConsumer<QueryExecutionResponseStream, QueryExecutionProgress> | null = null;
85 : /// The abort signal of the stream while it produces
86 1 : produceAbort: AbortSignal | null = null;
87 : /// The progress updates
88 1 : latestQueryProgress: QueryExecutionProgress | null;
89 : /// The metrics
90 1 : queryMetrics: QueryExecutionMetrics;
91 1 : endpoint: TrinoApiEndpoint;
92 :
93 : /// The constructor
94 1 : constructor(logger: Logger, apiClient: TrinoApiClientInterface, endpoint: TrinoApiEndpoint, result: TrinoQueryResult, metrics: QueryExecutionMetrics) {
95 1 : this.logger = logger;
96 1 : this.apiClient = apiClient;
97 1 : this.currentStatus = QueryExecutionStatus.RUNNING;
98 1 : this.responseMetadata = new Map();
99 1 : this.resultSchema = new AsyncValue();
100 1 : this.latestQueryResult = result;
101 1 : this.latestQueryStats = result.stats ?? null;
102 1 : this.latestQueryState = result.stats?.state ?? null;
103 1 : this.latestQueryReceivedAt = Date.now();
104 1 : this.latestQueryProgress = null;
105 1 : this.queryMetrics = metrics;
106 1 : this.endpoint = endpoint;
107 1 : if (result.columns) {
108 1 : this.resultSchema.resolve(translateTrinoSchema(result, this.logger));
109 : }
110 1 : }
111 :
112 : /// Cancel the query, a query without a next URI is done already
113 1 : async cancel(): Promise<void> {
114 1 : const nextUri = this.latestQueryResult?.nextUri;
115 1 : if (nextUri != null) await this.apiClient.cancelQuery(this.endpoint, nextUri);
116 1 : }
117 :
118 : /// Is the profile of the query shown? The plan and the operator summaries are fetched only then.
119 1 : setProfileDetail(wanted: boolean): void {
120 1 : this.profileDetail = wanted;
121 1 : if (!wanted) this.clearQueryInfoRetry();
122 1 : this.requestProfile();
123 1 : }
124 :
125 : /// Request the QueryInfo with the plans of the stages once the stages exist and the profile is shown. Trino
126 : /// reports the operator summaries of a stage (the rows of the plan) once its tasks are done, so QueryInfo is
127 : /// fetched again whenever the statement stats show more such stages. The requests do not hold up the query,
128 : /// a failed request leaves the stages without plan or rows. Pruned QueryInfo is no use here: it keeps only the
129 : /// output stage.
130 1 : requestProfile(): void {
131 1 : const progress = this.progressConsumer;
132 1 : const abort = this.produceAbort ?? undefined;
133 1 : const queryId = this.latestQueryResult?.id;
134 1 : if (!this.profileDetail || progress == null || this.latestQueryStats?.rootStage == null || !queryId) return;
135 1 : const doneStages = countDoneStages(this.latestQueryStats.rootStage);
136 1 : if (!this.queryInfoRequested) {
137 1 : this.queryInfoRequested = true;
138 1 : this.queryInfoDoneStages = doneStages;
139 1 : this.queryInfoPlan = this.apiClient.getQueryInfo(this.endpoint, queryId, abort).then(
140 1 : info => {
141 1 : progress.resolve(this, {
142 1 : ...deriveProgress(this.latestQueryStats, this.queryMetrics, this.latestQueryReceivedAt),
143 1 : engineProgress: { kind: 'trino_query_info', info, receivedAt: Date.now() },
144 1 : });
145 1 : this.retryQueryInfoIfIncomplete(info);
146 1 : },
147 0 : (e: unknown) => this.logger.warn("Failed to fetch the query plan", { "error": String(e), ...(e instanceof LoggableException ? e.keyValues : {}) }, LOG_CTX),
148 1 : );
149 1 : } else if (doneStages > this.queryInfoDoneStages) {
150 1 : this.queryInfoDoneStages = doneStages;
151 1 : void this.fetchQueryInfoStats();
152 1 : }
153 1 : }
154 :
155 : /// Fetch QueryInfo for the operator summaries, after the plans. One request at a time, a request while one is
156 : /// pending fetches once more after it.
157 1 : fetchQueryInfoStats(): Promise<TrinoQueryInfo | null> {
158 1 : if (this.queryInfoStats != null) {
159 1 : this.queryInfoStale = true;
160 1 : return this.queryInfoStats;
161 1 : }
162 1 : const progress = this.progressConsumer;
163 1 : const queryId = this.latestQueryResult?.id;
164 1 : const abort = this.produceAbort ?? undefined;
165 1 : if (progress == null || !queryId) return Promise.resolve(null);
166 1 : this.clearQueryInfoRetry();
167 1 : const request = (this.queryInfoPlan ?? Promise.resolve())
168 1 : .then(() => this.apiClient.getQueryInfo(this.endpoint, queryId, abort))
169 1 : .then(
170 1 : info => {
171 1 : progress.resolve(this, {
172 1 : ...deriveProgress(this.latestQueryStats, this.queryMetrics, this.latestQueryReceivedAt),
173 1 : engineProgress: { kind: 'trino_query_info_stats', info, receivedAt: Date.now() },
174 1 : });
175 1 : return info;
176 1 : },
177 0 : (e: unknown) => {
178 0 : this.logger.debug("Failed to fetch the query statistics", { "error": String(e), ...(e instanceof LoggableException ? e.keyValues : {}) }, LOG_CTX);
179 0 : return null;
180 0 : },
181 1 : )
182 1 : .finally(() => {
183 1 : this.queryInfoStats = null;
184 1 : });
185 1 : this.queryInfoStats = request;
186 1 : void request.then(info => {
187 1 : if (this.queryInfoStale) {
188 1 : this.queryInfoStale = false;
189 1 : void this.fetchQueryInfoStats();
190 1 : } else if (info != null) {
191 1 : this.retryQueryInfoIfIncomplete(info);
192 : }
193 1 : });
194 1 : return request;
195 1 : }
196 :
197 : /// Fetch QueryInfo again later while a stage with done tasks reports no operator summaries and the query runs
198 1 : retryQueryInfoIfIncomplete(info: TrinoQueryInfo): void {
199 1 : if (this.progressConsumer == null || !this.profileDetail || !lacksOperatorSummaries(info)) return;
200 1 : this.queryInfoRetry ??= setTimeout(() => {
201 1 : this.queryInfoRetry = null;
202 1 : void this.fetchQueryInfoStats();
203 1 : }, QUERY_INFO_RETRY_MS);
204 1 : }
205 :
206 : /// Clear the timer of the next QueryInfo
207 1 : clearQueryInfoRetry(): void {
208 1 : if (this.queryInfoRetry != null) clearTimeout(this.queryInfoRetry);
209 1 : this.queryInfoRetry = null;
210 1 : }
211 :
212 : /// Fetch the QueryInfo of a finished query until it is final: then the operator summaries of all stages are
213 : /// complete. The query waits for it, but not longer than a bound.
214 1 : async finishQueryInfo(finished: boolean): Promise<void> {
215 1 : this.clearQueryInfoRetry();
216 1 : if (finished && this.profileDetail && this.queryInfoRequested) {
217 1 : const deadline = Date.now() + QUERY_INFO_FINAL_TIMEOUT_MS;
218 1 : let timer: ReturnType<typeof setTimeout> | null = null;
219 1 : const bound = new Promise<void>(resolve => { timer = setTimeout(resolve, QUERY_INFO_FINAL_TIMEOUT_MS); });
220 1 : const final = (async () => {
221 : // A pending request may predate the end of the query
222 1 : await this.queryInfoStats;
223 1 : while (true) {
224 1 : const info = await this.fetchQueryInfoStats();
225 1 : if (info == null || isFinalQueryInfo(info) || Date.now() + QUERY_INFO_FINAL_RETRY_MS >= deadline) break;
226 1 : await new Promise(resolve => setTimeout(resolve, QUERY_INFO_FINAL_RETRY_MS));
227 1 : }
228 1 : })();
229 1 : await Promise.race([final, bound]);
230 1 : if (timer != null) clearTimeout(timer);
231 1 : }
232 : // Progress after the end of the stream is dropped
233 1 : this.progressConsumer = null;
234 1 : this.produceAbort = null;
235 1 : this.clearQueryInfoRetry();
236 1 : }
237 :
238 : /// Fetch the next query result
239 1 : async fetchNextQueryResult(abort?: AbortSignal): Promise<TrinoQueryResult | null> {
240 : // Do we have a next URI?
241 1 : const nextUri = this.latestQueryResult?.nextUri;
242 1 : if (!nextUri) {
243 1 : return null;
244 : }
245 1 : this.logger.debug("Fetching next query results", { "nextUri": nextUri }, LOG_CTX);
246 :
247 : // Get the next query result
248 1 : this.queryMetrics.totalQueryRequestsStarted += 1;
249 1 : const timeBefore = (new Date()).getTime();
250 1 : const queryResult = await this.apiClient.getQueryResult(this.endpoint, nextUri, abort);
251 1 : const timeAfter = (new Date()).getTime();
252 1 : this.latestQueryResult = queryResult;
253 1 : this.latestQueryStats = queryResult.stats ?? null;
254 1 : this.latestQueryState = this.latestQueryStats?.state ?? null;
255 1 : this.latestQueryReceivedAt = Date.now();
256 :
257 : // Did the query fail?
258 1 : if (queryResult.error) {
259 1 : const errorCode = queryResult.error.errorCode;
260 1 : const errorName = queryResult.error.errorName;
261 1 : const errorType = queryResult.error.errorType;
262 1 : const errorMessage = queryResult.error.message;
263 :
264 1 : const rawError: RawProxyError = {
265 1 : message: "Query returned an error",
266 1 : data: {
267 1 : "errorCode": errorCode.toString(),
268 1 : "errorName": errorName,
269 1 : "errorType": errorType,
270 1 : "errorMessage": errorMessage,
271 1 : },
272 1 : };
273 1 : const error = new ChannelError(rawError, errorCode, undefined, LOG_CTX);
274 1 : if (!this.resultSchema.isResolved()) {
275 1 : this.resultSchema.reject(error);
276 : }
277 1 : this.queryMetrics.totalQueryRequestsFailed += 1;
278 1 : throw error;
279 1 : } else {
280 1 : this.queryMetrics.totalQueryRequestsSucceeded += 1;
281 : }
282 1 : this.queryMetrics.totalQueryRequestDurationMs += timeAfter - timeBefore;
283 :
284 : // Do we already have a schema?
285 1 : if (!this.resultSchema.isResolved() && queryResult.columns) {
286 : // Attempt to translate the schema
287 1 : const translated = translateTrinoSchema(queryResult, this.logger);
288 1 : this.resultSchema.resolve(translated);
289 1 : }
290 1 : return queryResult;
291 1 : }
292 :
293 : /// Get the result metadata (after completion)
294 0 : getMetadata(): Map<string, string> {
295 0 : return new Map();
296 0 : }
297 : /// Get the stream metrics
298 1 : getMetrics(): QueryExecutionMetrics {
299 1 : return this.queryMetrics;
300 1 : }
301 : /// Get the current query status
302 0 : getStatus(): QueryExecutionStatus {
303 0 : return this.currentStatus;
304 0 : }
305 : /// Await the schema message
306 1 : async getSchema(): Promise<arrow.Schema | null> {
307 1 : return this.resultSchema.getValue();
308 1 : }
309 : /// Await the next record batch.
310 : /// The stream follows the next URIs until Trino stops sending one, the query state does not end the stream:
311 : /// Trino reports states like PLANNING before the rows and FINISHED while the coordinator still buffers rows.
312 1 : async produce(batches: AsyncConsumer<QueryExecutionResponseStream, arrow.RecordBatch | null>, progress: AsyncConsumer<QueryExecutionResponseStream, QueryExecutionProgress>, abort?: AbortSignal): Promise<void> {
313 1 : this.progressConsumer = progress;
314 1 : this.produceAbort = abort ?? null;
315 : // The response to the query request carries the first progress (QUEUED)
316 1 : if (this.latestQueryStats != null) {
317 1 : this.latestQueryProgress = deriveProgress(this.latestQueryStats, this.queryMetrics, this.latestQueryReceivedAt);
318 1 : progress.resolve(this, this.latestQueryProgress);
319 1 : this.requestProfile();
320 1 : }
321 1 : let pending = this.fetchNextQueryResult(abort);
322 1 : try {
323 1 : while (true) {
324 1 : const result = await pending;
325 1 : abort?.throwIfAborted();
326 1 : if (result == null) {
327 1 : break;
328 : }
329 : // Request the next page before translating this one, Trino fills the page while we translate
330 1 : pending = this.fetchNextQueryResult(abort);
331 :
332 : // Has a data attribute?
333 1 : if (Array.isArray(result.data)) {
334 : // Have result data but not schema yet?
335 : // This is unexpected.
336 1 : if (!this.resultSchema.isResolved()) {
337 1 : throw new Error("Result schema is missing");
338 : }
339 : // Translate the trino batch
340 1 : const schema = this.resultSchema.getResolvedValue();
341 1 : const resultBatch = translateAnyRowsToArrowBatch(schema!, result.data, this.logger);
342 1 : this.queryMetrics.totalBatchesReceived += 1;
343 1 : this.queryMetrics.totalRowsReceived += resultBatch.numRows;
344 :
345 : // Publish a new progress update
346 1 : this.latestQueryProgress = deriveProgress(result.stats ?? null, this.queryMetrics, this.latestQueryReceivedAt);
347 1 : progress.resolve(this, this.latestQueryProgress);
348 1 : this.requestProfile();
349 :
350 : // Produce a batch
351 1 : batches.resolve(this, resultBatch);
352 1 : abort?.throwIfAborted();
353 1 : await yieldToBrowser();
354 1 : } else {
355 : // Publish the progress update
356 1 : this.latestQueryProgress = deriveProgress(result.stats ?? null, this.queryMetrics, this.latestQueryReceivedAt);
357 1 : progress.resolve(this, this.latestQueryProgress);
358 1 : this.requestProfile();
359 1 : }
360 1 : }
361 1 : this.logger.debug("Reached end of query result stream", { "queryState": this.latestQueryState }, LOG_CTX);
362 1 : await this.finishQueryInfo(true);
363 1 : } catch (e: any) {
364 : // Don't leave the request of the next page as unhandled rejection
365 1 : pending.catch(() => { });
366 1 : await this.finishQueryInfo(false);
367 1 : throw e;
368 1 : }
369 1 : }
370 1 : }
371 :
372 : export interface TrinoChannelInterface {
373 : /// Execute Query
374 : executeQuery(param: proto.salesforce_hyperdb_grpc_v1.pb.QueryParam, abort?: AbortSignal): Promise<TrinoQueryResultStream>;
375 : /// Destroy the connection
376 : close(): Promise<void>;
377 : }
378 :
379 1 : export class TrinoChannel implements TrinoChannelInterface {
380 : /// The logger
381 1 : logger: Logger;
382 : /// The client
383 1 : apiClient: TrinoApiClientInterface;
384 : /// The trino api endpoint
385 1 : endpoint: TrinoApiEndpoint;
386 : /// The catalog and schema of the names that statements leave out
387 1 : defaults: TrinoSessionDefaults;
388 :
389 : /// Constructor
390 1 : constructor(logger: Logger, client: TrinoApiClientInterface, endpoint: TrinoApiEndpoint, defaults: TrinoSessionDefaults) {
391 1 : this.logger = logger;
392 1 : this.apiClient = client;
393 1 : this.endpoint = endpoint;
394 1 : this.defaults = defaults;
395 1 : }
396 :
397 : /// Execute Query
398 1 : async executeQuery(param: proto.salesforce_hyperdb_grpc_v1.pb.QueryParam, abort?: AbortSignal): Promise<TrinoQueryResultStream> {
399 1 : const metrics = createQueryResponseStreamMetrics();
400 1 : const timeBefore = (new Date()).getTime();
401 :
402 1 : try {
403 1 : this.logger.debug("Executing query", {}, LOG_CTX);
404 1 : metrics.totalQueryRequestsStarted += 1;
405 :
406 1 : const result = await this.apiClient.runQuery(this.endpoint, this.defaults, param.query, abort);
407 1 : const timeAfter = (new Date()).getTime();
408 :
409 1 : metrics.totalQueryRequestsSucceeded += 1;
410 1 : metrics.totalQueryRequestDurationMs += timeAfter - timeBefore;
411 :
412 1 : this.logger.debug("Opened query result stream", {}, LOG_CTX);
413 1 : const stream = new TrinoQueryResultStream(this.logger, this.apiClient, this.endpoint, result, metrics);
414 1 : return stream;
415 1 : } catch (e: any) {
416 1 : const timeAfter = (new Date()).getTime();
417 1 : metrics.totalQueryRequestsFailed += 1;
418 1 : metrics.totalQueryRequestDurationMs += timeAfter - timeBefore;
419 1 : throw e;
420 1 : }
421 1 : }
422 :
423 : /// Destroy the connection
424 0 : async close(): Promise<void> {
425 : return;
426 0 : }
427 1 : }
428 :
429 : /// Parse a Trino type string and return the corresponding Arrow DataType
430 1 : function parseTrinoType(typeStr: string): arrow.DataType {
431 1 : const type = typeStr.toLowerCase().trim();
432 :
433 : // Boolean
434 1 : if (type === "boolean") {
435 1 : return new arrow.Bool();
436 : }
437 : // Integer types
438 1 : if (type === "tinyint") {
439 1 : return new arrow.Int8();
440 : }
441 1 : if (type === "smallint") {
442 1 : return new arrow.Int16();
443 : }
444 1 : if (type === "integer" || type === "int") {
445 1 : return new arrow.Int32();
446 : }
447 1 : if (type === "bigint") {
448 1 : return new arrow.Int64();
449 : }
450 : // Floating-point types
451 1 : if (type === "real") {
452 1 : return new arrow.Float32();
453 : }
454 1 : if (type === "double") {
455 1 : return new arrow.Float64();
456 : }
457 :
458 : // Decimal types: decimal(precision, scale)
459 1 : const decimalMatch = type.match(/^decimal\((\d+),\s*(\d+)\)$/);
460 1 : if (decimalMatch) {
461 1 : const precision = parseInt(decimalMatch[1], 10);
462 1 : const scale = parseInt(decimalMatch[2], 10);
463 1 : return new arrow.Decimal(scale, precision, 128);
464 1 : }
465 1 : if (type === "decimal") {
466 : // Default decimal without precision/scale
467 1 : return new arrow.Decimal(0, 38, 128);
468 : }
469 :
470 : // String types
471 1 : if (type === "varchar" || type.startsWith("varchar(") || type === "char" || type.startsWith("char(")) {
472 1 : return new arrow.Utf8();
473 : }
474 1 : if (type === "varbinary") {
475 1 : return new arrow.Binary();
476 : }
477 1 : if (type === "json") {
478 1 : return new arrow.Utf8(); // JSON is represented as string
479 : }
480 : // Date/Time types
481 1 : if (type === "date") {
482 1 : return new arrow.DateDay();
483 : }
484 1 : if (type === "time" || type.startsWith("time(")) {
485 1 : return new arrow.TimeMillisecond();
486 : }
487 1 : if (type === "time with time zone" || type.startsWith("time(") && type.includes("with time zone")) {
488 0 : return new arrow.TimeMillisecond();
489 : }
490 1 : if (type === "timestamp" || type.startsWith("timestamp(")) {
491 1 : if (type.includes("with time zone")) {
492 1 : return new arrow.TimestampMillisecond("UTC");
493 : }
494 1 : return new arrow.TimestampMillisecond();
495 1 : }
496 : // Interval types - represent as strings since Arrow has limited interval support
497 1 : if (type.startsWith("interval")) {
498 1 : return new arrow.Utf8();
499 : }
500 : // UUID - represent as string
501 1 : if (type === "uuid") {
502 1 : return new arrow.Utf8();
503 : }
504 : // IP address - represent as string
505 1 : if (type === "ipaddress") {
506 1 : return new arrow.Utf8();
507 : }
508 : // Complex types (array, map, row) - represent as JSON strings for simplicity
509 : // The Arrow JS API for complex types is cumbersome; JSON serialization is practical
510 1 : if (type.startsWith("array(") || type.startsWith("map(") || type.startsWith("row(")) {
511 1 : return new arrow.Utf8();
512 : }
513 :
514 : // Unknown type - fall back to string representation
515 1 : return new arrow.Utf8();
516 1 : }
517 :
518 : /// Translate the Trino schema
519 1 : function translateTrinoSchema(result: TrinoQueryResult, logger: Logger): arrow.Schema {
520 1 : let fields: arrow.Field[] = [];
521 1 : for (const column of result.columns!) {
522 1 : const arrowType = parseTrinoType(column.type);
523 1 : fields.push(new arrow.Field(column.name, arrowType, true));
524 1 : }
525 1 : return new arrow.Schema(fields);
526 1 : }
527 :
528 : /// Derive the execution progress of the statistics of a response that arrived at a time (milliseconds since epoch)
529 1 : export function deriveProgress(stats: TrinoQueryStatistics | null, metrics: QueryExecutionMetrics, receivedAt: number): QueryExecutionProgress {
530 1 : return {
531 1 : isQueued: stats?.queued ?? null,
532 1 : progressPercent: stats?.progressPercentage ?? null,
533 1 : resultRows: null,
534 1 : rowsProcessed: stats?.processedRows ?? null,
535 1 : executionTimeMs: stats?.elapsedTimeMillis ?? null,
536 1 : metrics: { ...metrics },
537 1 : phase: stats?.state ?? null,
538 1 : engineProgress: stats == null ? null : { kind: 'trino_statement_stats', stats, receivedAt },
539 1 : };
540 1 : }
|
