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