Line data Source code
1 : import * as React from 'react';
2 :
3 : import { useDynamicConnectionDispatch } from '../connection_registry.js';
4 : import { CatalogUpdateTaskState, CatalogUpdateTaskStatus, CatalogUpdateVariant, MAX_TOUCHED_ATTEMPTS } from './catalog_update_state.js';
5 : import { useSalesforceAPI } from '../salesforce/salesforce_connector.js';
6 : import { HYPER_CONNECTOR, SALESFORCE_DATA_CLOUD_CONNECTOR, DBCOM_CONNECTOR, TRINO_CONNECTOR } from '../connector_info.js';
7 : import {
8 : ConnectionHealth,
9 : CATALOG_UPDATE_CANCELLED,
10 : CATALOG_UPDATE_FAILED,
11 : CATALOG_UPDATE_PARTIALLY_SUCCEEDED,
12 : CATALOG_UPDATE_SUCCEEDED,
13 : UPDATE_CATALOG,
14 : } from '../connection_state.js';
15 : import { getHyperCloudAttachedDatabase, updateSalesforceCatalog } from '../salesforce/salesforce_catalog_update.js';
16 : import { useQueryExecutor } from '../query_executor.js';
17 : import { useLogger } from '../../platform/logger/logger_provider.js';
18 : import { createTrace } from '../../platform/logger/trace_context.js';
19 : import { updateInformationSchemaCatalog, updateInformationSchemaRelations } from './catalog_query_information_schema.js';
20 : import type { RelationName } from './catalog_sql_generator.js';
21 : import { queryTrinoCatalogScopes, trinoCatalogScopes } from '../trino/trino_catalogs.js';
22 : import { updateHyperCatalog } from '../hyper/hyper_catalog_update.js';
23 : import { connectionCatalogIsDurable } from '../connection_params.js';
24 : import { ensureConnectionClientFunctions } from '../connection_client_functions.js';
25 : import { connectorOfDetails } from '../connector_descriptor.js';
26 :
27 1 : const LOG_CTX = 'catalog_loader';
28 :
29 : /// The wait before asking a connection again about the relations of executed statements, per failed attempt
30 1 : const TOUCHED_RETRY_DELAY_MS = 2000;
31 1 : let NEXT_CATALOG_UPDATE_ID = 1;
32 :
33 : /// The catalog update args
34 : interface CatalogLoaderArgs {
35 : /// The relations to ask the connection about, everything if absent
36 : relations: RelationName[] | null;
37 : }
38 : /// The catalog updater function
39 : export type CatalogLoader = (connectionId: string, args: CatalogLoaderArgs) => [number, Promise<void>];
40 : /// A function to add a connection id to a catalog loader queue
41 : type RefreshCatalogFn = (connectionId: string, force: boolean) => void;
42 : /// A catalog loader queue
43 : interface CatalogLoaderQueue { queue: Map<string, boolean>; }
44 :
45 : /// The React context to propagate a active catalog loader queue function
46 1 : const LOADER_QUEUE_FN_CTX = React.createContext<RefreshCatalogFn | null>(null);
47 :
48 : /// The hook to resolve the catalog queue
49 1 : export const useCatalogLoaderQueue = () => React.useContext(LOADER_QUEUE_FN_CTX)!;
50 :
51 : export interface CatalogLoaderDependencies {
52 : useDynamicConnectionDispatch: typeof useDynamicConnectionDispatch;
53 : useQueryExecutor: typeof useQueryExecutor;
54 : useSalesforceAPI: typeof useSalesforceAPI;
55 : updateHyperCatalog: typeof updateHyperCatalog;
56 : }
57 :
58 1 : const DEFAULT_DEPENDENCIES: CatalogLoaderDependencies = {
59 1 : useDynamicConnectionDispatch,
60 1 : useQueryExecutor,
61 1 : useSalesforceAPI,
62 1 : updateHyperCatalog,
63 1 : };
64 :
65 : /// The props of the catalog loader
66 : export interface CatalogLoaderProviderProps {
67 : /// The children
68 : children?: React.ReactElement;
69 : /// The hooks the loader uses, replaced in tests
70 : dependencies?: CatalogLoaderDependencies;
71 : /// Called when the catalog of a connection changed
72 : onCatalogDidUpdate?: (connectionId: string) => void;
73 : }
74 :
75 1 : export function CatalogLoaderProvider(props: CatalogLoaderProviderProps) {
76 1 : const dependencies = props.dependencies ?? DEFAULT_DEPENDENCIES;
77 1 : const logger = useLogger();
78 1 : const executor = dependencies.useQueryExecutor();
79 1 : const sfapi = dependencies.useSalesforceAPI();
80 :
81 : // The connection registry changes frequently, the connection map is stable.
82 : // This executor will depend on the map directly since it can resolve everything ad-hoc.
83 1 : const [connReg, connDispatch] = dependencies.useDynamicConnectionDispatch();
84 1 : const connMap = connReg.connections;
85 1 : const onCatalogDidUpdate = React.useRef(props.onCatalogDidUpdate);
86 1 : onCatalogDidUpdate.current = props.onCatalogDidUpdate;
87 :
88 : // The client functions of a connection come with its connector, scripts that call them are re-analyzed
89 1 : React.useEffect(() => {
90 1 : for (const connectionId of connMap.keys()) {
91 1 : ensureConnectionClientFunctions(connMap, connectionId, id => onCatalogDidUpdate.current?.(id), logger);
92 : }
93 1 : }, [connReg, connMap, logger]);
94 :
95 : // Maintain a queue
96 1 : const [queueState, setQueueState] = React.useState<CatalogLoaderQueue>(() => ({ queue: new Map() }));
97 1 : const enqueue = React.useCallback<RefreshCatalogFn>((connectionId: string, force: boolean) => {
98 1 : setQueueState(s => ({ queue: new Map(s.queue).set(connectionId, force || s.queue.get(connectionId) === true) }));
99 1 : }, []);
100 :
101 : // Execute a query with pre-allocated query id
102 1 : const updateImpl = React.useCallback(async (connectionId: string, args: CatalogLoaderArgs, updateId: number): Promise<void> => {
103 : // Each catalog update gets its own trace so failures don't borrow
104 : // whatever trace happens to be on a shared stack.
105 1 : const traced = logger.withTrace(createTrace());
106 :
107 1 : const conn = connMap.get(connectionId);
108 1 : if (!conn) {
109 1 : traced.warn("Failed to resolve connection", { connectionId }, LOG_CTX);
110 1 : throw new Error(`couldn't find a connection with id ${connectionId}`);
111 1 : }
112 1 : if (!executor) {
113 1 : traced.warn("Query executor not configured", { connectionId }, LOG_CTX);
114 1 : throw new Error(`couldn't find trino executor`);
115 1 : }
116 :
117 1 : traced.debug("Updating catalog", { connectionId }, LOG_CTX);
118 :
119 1 : const relations = args.relations;
120 :
121 : // Accept the query and clear the request
122 1 : const abortController = new AbortController();
123 1 : const now = new Date();
124 1 : const initialState: CatalogUpdateTaskState = {
125 1 : taskId: updateId,
126 1 : taskVariant: relations != null ? CatalogUpdateVariant.RELATION_REFRESH : CatalogUpdateVariant.FULL_CATALOG_REFRESH,
127 1 : status: CatalogUpdateTaskStatus.STARTED,
128 1 : cancellation: abortController,
129 1 : queries: [],
130 1 : error: null,
131 1 : startedAt: now,
132 1 : finishedAt: null,
133 1 : touchedRelations: relations ?? [],
134 1 : };
135 1 : connDispatch(connectionId, {
136 1 : type: UPDATE_CATALOG,
137 1 : value: [updateId, initialState],
138 1 : });
139 :
140 : // Emit a heartbeat log every few seconds so long-running updates
141 : // don't look like a silent hang between "Starting" and "Updated".
142 1 : const updateStartMs = performance.now();
143 1 : const heartbeat = setInterval(() => {
144 0 : const elapsedMs = performance.now() - updateStartMs;
145 0 : traced.info("Catalog update in progress", {
146 0 : connectionId,
147 0 : "updateId": updateId.toString(),
148 0 : "elapsedMs": elapsedMs.toFixed(0),
149 0 : }, LOG_CTX);
150 1 : }, 5000);
151 :
152 : // Update the catalog
153 1 : const versionBefore = conn.catalog.version();
154 1 : try {
155 1 : const batchOptions = {
156 1 : abortSignal: abortController.signal,
157 1 : isCurrent: () => connMap.get(connectionId)?.catalog === conn.catalog,
158 1 : };
159 1 : let partialError: Error | null = null;
160 1 : switch (conn.details.type) {
161 : // Update the catalog by querying the information_schema
162 1 : case TRINO_CONNECTOR: {
163 : // Trino folds the names of relations to lower case
164 1 : if (relations != null) {
165 1 : const folded = relations.map(r => ({ databaseName: r.databaseName.toLowerCase(), schemaName: r.schemaName.toLowerCase(), tableName: r.tableName.toLowerCase() }));
166 1 : await updateInformationSchemaRelations(connectionId, connDispatch, updateId, folded, executor, conn.catalog, conn.catalogRelationEntryId, batchOptions);
167 1 : break;
168 1 : }
169 1 : const setupParams = conn.details.value.proto.setupParams;
170 1 : let scopes = setupParams ? trinoCatalogScopes(setupParams) : [];
171 : // Without configured catalogs, read all catalogs of the server
172 1 : if (scopes.length === 0) {
173 1 : scopes = await queryTrinoCatalogScopes(connectionId, connDispatch, updateId, executor, abortController.signal);
174 : }
175 1 : await updateInformationSchemaCatalog(connectionId, connDispatch, updateId, scopes, executor, conn.catalog, conn.catalogRelationEntryId, batchOptions);
176 1 : break;
177 1 : }
178 1 : case HYPER_CONNECTOR: {
179 1 : const result = await dependencies.updateHyperCatalog(
180 1 : traced,
181 1 : connectionId,
182 1 : connDispatch,
183 1 : updateId,
184 1 : (conn.details.value.proto.setupParams?.attachedDatabases ?? []).map(database => ({
185 0 : path: database.path ?? '',
186 0 : alias: database.alias,
187 1 : })),
188 1 : executor,
189 1 : conn.catalog,
190 1 : conn.instance,
191 1 : conn.catalogRelationEntryId,
192 1 : conn.catalogFunctionScript,
193 1 : conn.details.value.proto.setupParams?.protocol === 'WASM',
194 1 : abortController.signal,
195 1 : batchOptions,
196 1 : relations,
197 1 : );
198 1 : if (result.failures.length > 0) {
199 1 : partialError = new Error(result.failures
200 1 : .map(failure => `${failure.database}: ${failure.error.message}`)
201 1 : .join('; '));
202 1 : traced.warn('Partially updated Hyper catalog', {
203 1 : connectionId,
204 1 : updated: result.updatedDatabases.join(', '),
205 1 : failed: result.failures.map(failure => failure.database).join(', '),
206 1 : }, LOG_CTX);
207 1 : }
208 1 : break;
209 1 : }
210 : // Update the catalog by querying the _hyper_catalog of the attached Hyper Cloud database
211 1 : case DBCOM_CONNECTOR:
212 1 : await dependencies.updateHyperCatalog(
213 1 : traced,
214 1 : connectionId,
215 1 : connDispatch,
216 1 : updateId,
217 1 : [getHyperCloudAttachedDatabase(conn.details.value.proto.setupParams)],
218 1 : executor,
219 1 : conn.catalog,
220 1 : conn.instance,
221 1 : conn.catalogRelationEntryId,
222 1 : conn.catalogFunctionScript,
223 1 : false,
224 1 : abortController.signal,
225 1 : batchOptions,
226 1 : relations,
227 1 : );
228 1 : break;
229 : // Update the catalog by querying the Salesforce Metadata Service
230 1 : case SALESFORCE_DATA_CLOUD_CONNECTOR:
231 1 : await updateSalesforceCatalog(
232 1 : traced,
233 1 : conn.details.value,
234 1 : conn.catalog,
235 1 : conn.instance,
236 1 : conn.catalogRelationEntryId,
237 1 : conn.catalogFunctionScript,
238 1 : sfapi,
239 1 : abortController,
240 1 : batchOptions,
241 1 : );
242 : break;
243 1 : }
244 1 : traced.debug("Updated catalog", { connectionId }, LOG_CTX);
245 :
246 : // Mark the update successful
247 1 : if (partialError == null) {
248 1 : connDispatch(connectionId, {
249 1 : type: CATALOG_UPDATE_SUCCEEDED,
250 1 : value: [updateId, conn.catalog.version() !== versionBefore],
251 1 : });
252 : } else {
253 1 : connDispatch(connectionId, {
254 1 : type: CATALOG_UPDATE_PARTIALLY_SUCCEEDED,
255 1 : value: [updateId, partialError],
256 1 : });
257 : }
258 : // Announce the new catalog
259 1 : onCatalogDidUpdate.current?.(connectionId);
260 :
261 1 : } catch (e: any) {
262 1 : if (e?.name === 'AbortError') {
263 1 : traced.info("Cancelled catalog update", { connectionId, "error": e?.message ?? String(e) }, LOG_CTX);
264 1 : connDispatch(connectionId, {
265 1 : type: CATALOG_UPDATE_CANCELLED,
266 1 : value: [updateId, e],
267 1 : });
268 1 : } else {
269 1 : traced.error("Failed to update catalog", { connectionId, "error": e?.message ?? String(e) }, LOG_CTX);
270 1 : connDispatch(connectionId, {
271 1 : type: CATALOG_UPDATE_FAILED,
272 1 : value: [updateId, e],
273 1 : });
274 1 : }
275 1 : } finally {
276 1 : clearInterval(heartbeat);
277 1 : const totalMs = performance.now() - updateStartMs;
278 1 : traced.info("Finished catalog update", {
279 1 : connectionId,
280 1 : "updateId": updateId.toString(),
281 1 : "durationMs": totalMs.toFixed(0),
282 1 : }, LOG_CTX);
283 1 : }
284 1 : }, [connMap, sfapi, executor, dependencies.updateHyperCatalog]);
285 :
286 : // Allocate the next query id and start the execution
287 1 : const update = React.useCallback<CatalogLoader>((connectionId: string, args: CatalogLoaderArgs): [number, Promise<void>] => {
288 1 : const updateId = NEXT_CATALOG_UPDATE_ID++;
289 1 : const execution = updateImpl(connectionId, args, updateId);
290 1 : return [updateId, execution];
291 1 : }, [updateImpl]);
292 :
293 : // A full refresh only runs on user input and for the initial description of a connection: the catalog of a
294 : // connection whose database doesn't outlive the app is not restored from storage, it is described whenever the
295 : // connection comes online. A connection that has neither a stored nor a refreshed catalog is described when it
296 : // first comes online. Everything else asks the connection about single relations.
297 1 : const onlineRef = React.useRef(new Set<string>());
298 1 : React.useEffect(() => {
299 1 : const online = onlineRef.current;
300 1 : for (const [connectionId, conn] of connMap) {
301 1 : if (conn.connectionHealth !== ConnectionHealth.ONLINE) {
302 1 : online.delete(connectionId);
303 1 : } else if (!online.has(connectionId)) {
304 1 : online.add(connectionId);
305 1 : if (!connectionCatalogIsDurable(conn.params)) {
306 1 : enqueue(connectionId, true);
307 1 : } else if (!conn.catalogUpdates.described) {
308 1 : enqueue(connectionId, false);
309 : }
310 1 : }
311 : }
312 1 : for (const connectionId of [...online]) {
313 1 : if (!connMap.has(connectionId)) online.delete(connectionId);
314 : }
315 1 : }, [connReg, connMap, enqueue]);
316 :
317 : // When the last update of a connection finished and the timer of the next attempt
318 1 : const lastFinishedAt = React.useRef(new Map<string, number>());
319 1 : const retryTimer = React.useRef<ReturnType<typeof setTimeout> | null>(null);
320 1 : React.useEffect(() => () => {
321 1 : if (retryTimer.current != null) clearTimeout(retryTimer.current);
322 1 : retryTimer.current = null;
323 1 : }, []);
324 :
325 : // Subscribe the queue
326 1 : const updatesInProgress = React.useRef<Set<string> | null>(null);
327 1 : React.useEffect(() => {
328 1 : const inProgress = updatesInProgress.current ?? new Set();
329 1 : updatesInProgress.current = inProgress;
330 :
331 : // Helper to perform the catalog update
332 1 : const doUpdate = async (connectionId: string, relations: RelationName[] | null) => {
333 1 : inProgress.add(connectionId);
334 1 : try {
335 1 : logger.info("Starting catalog update", { connectionId, relations: String(relations?.length ?? 'all') }, LOG_CTX);
336 : // Otherwise start the catalog update
337 1 : const [_updateId, updatePromise] = update(connectionId, { relations });
338 : // Await the update
339 1 : await updatePromise;
340 1 : } catch (e: any) {
341 1 : logger.warn("Catalog update failed", { connectionId, "error": e?.message ?? String(e) }, LOG_CTX);
342 1 : } finally {
343 1 : inProgress.delete(connectionId);
344 1 : lastFinishedAt.current.set(connectionId, Date.now());
345 : // Pending requests must be reconsidered even if the registry does not change.
346 1 : setQueueState(s => ({ ...s }));
347 1 : }
348 1 : };
349 :
350 : // Ask the connections about the relations of executed statements, when they have room for another query.
351 : // A refresh that runs may have read the relations too early, this update follows it and corrects them.
352 1 : for (const [connectionId, conn] of connReg.connections) {
353 1 : const { touchedRelations: relations, touchedFailures: failures, tasksRunning } = conn.catalogUpdates;
354 1 : if (relations.length === 0 || inProgress.has(connectionId) || tasksRunning.size > 0) continue;
355 : // A requested refresh goes first, the lookup follows it
356 1 : if (queueState.queue.has(connectionId)) continue;
357 : // Relations that a restored catalog still has to ask about wait for the connection
358 1 : if (conn.connectionHealth !== ConnectionHealth.ONLINE) continue;
359 1 : if (conn.queriesActive.size >= connectorOfDetails(conn.details).maxConcurrentQueries(conn.details.value)) continue;
360 : // The next statement or a manual refresh tries again
361 1 : if (failures >= MAX_TOUCHED_ATTEMPTS) continue;
362 1 : if (failures > 0) {
363 1 : const wait = (lastFinishedAt.current.get(connectionId) ?? 0) + TOUCHED_RETRY_DELAY_MS * failures - Date.now();
364 1 : if (wait > 0) {
365 1 : if (retryTimer.current == null) {
366 1 : retryTimer.current = setTimeout(() => {
367 1 : retryTimer.current = null;
368 1 : setQueueState(s => ({ ...s }));
369 1 : }, wait);
370 : }
371 1 : continue;
372 1 : }
373 1 : }
374 1 : doUpdate(connectionId, relations);
375 1 : }
376 :
377 1 : const processed: string[] = [];
378 1 : for (const [connectionId, force] of queueState.queue) {
379 1 : logger.debug("Received catalog update request", { connectionId }, LOG_CTX);
380 :
381 : // Find the connection
382 1 : const connState = connReg.connections.get(connectionId);
383 1 : if (!connState) {
384 1 : logger.warn("Failed to resolve connection", { connectionId }, LOG_CTX);
385 1 : processed.push(connectionId);
386 1 : continue;
387 1 : }
388 :
389 : // Has a current catalog update running?
390 : // Skip auto-updates, but let explicit user refreshes supersede
391 : // the in-flight ones by aborting them first.
392 1 : if (connState.catalogUpdates.tasksRunning.size > 0) {
393 1 : if (!force) {
394 1 : logger.info("Skipping redundant catalog update", { connectionId }, LOG_CTX);
395 1 : continue;
396 1 : }
397 1 : for (const task of connState.catalogUpdates.tasksRunning.values()) {
398 1 : if (!task.cancellation.signal.aborted) {
399 1 : task.cancellation.abort(new DOMException('Superseded by forced catalog refresh', 'AbortError'));
400 : }
401 : }
402 1 : }
403 :
404 : // Keep forced requests queued until the cancelled execution settles.
405 1 : if (inProgress.has(connectionId)) {
406 1 : continue;
407 : }
408 :
409 : // A refresh described the connection meanwhile
410 1 : if (!force && connState.catalogUpdates.described) {
411 1 : processed.push(connectionId);
412 1 : continue;
413 1 : }
414 :
415 : // Perform the catalog update
416 1 : doUpdate(connectionId, null);
417 1 : processed.push(connectionId);
418 1 : }
419 :
420 : // No processed?
421 1 : if (processed.length == 0) {
422 1 : return;
423 : }
424 :
425 : // Remove all processed ids from the queue
426 1 : setQueueState(s => {
427 1 : const queue = new Map(s.queue);
428 1 : for (const connectionId of processed) {
429 1 : queue.delete(connectionId);
430 : }
431 1 : return { ...s, queue };
432 1 : });
433 1 : }, [queueState, connReg, update, logger]);
434 :
435 1 : return (
436 1 : <LOADER_QUEUE_FN_CTX.Provider value={enqueue}>
437 1 : {props.children}
438 1 : </LOADER_QUEUE_FN_CTX.Provider>
439 : );
440 1 : }
|
