feat(metadata): share table metadata cache
This commit is contained in:
vrustx
GitHub
parent
81f787e8d1
commit
3da3849953
|
|
@ -4,19 +4,20 @@ import { useConnectionStore } from "@/stores/connectionStore";
|
|||
import { useQueryStore } from "@/stores/queryStore";
|
||||
import { buildTableSelectSql, quoteTableDataIdentifier } from "@/lib/table/tableSelectSql";
|
||||
import { tableOpenPageLimit } from "@/lib/table/tableOpenPageLimit";
|
||||
import { editableRowIdentifierColumns, usesSyntheticRowIdKey } from "@/lib/table/tableEditing";
|
||||
import { usesSyntheticRowIdKey } from "@/lib/table/tableEditing";
|
||||
import { tableMetaForDataTab } from "@/lib/table/tableDataTabMeta";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import type { QueryTab } from "@/types/database";
|
||||
import { useToast } from "@/composables/useToast";
|
||||
import { effectiveDatabaseTypeForConnection, metadataSchemaForConnection } from "@/lib/database/jdbcDialect";
|
||||
import { loadTableMetadata, TABLE_METADATA_CACHE_TTL_MS } from "@/lib/metadata/tableMetadataCache";
|
||||
import { applyMongoFindSort } from "@/lib/mongo/mongoShellCommand";
|
||||
import { uuid } from "@/lib/common/utils";
|
||||
import type { DataGridSortMode } from "@/lib/dataGrid/dataGridSort";
|
||||
import type { DataGridReloadIntent } from "@/lib/dataGrid/dataGridToolbar";
|
||||
import { queryResultBaseSql, queryResultExecutionSql } from "@/lib/tabs/tabPresentation";
|
||||
|
||||
const DATA_TAB_METADATA_TTL_MS = 30_000;
|
||||
const DATA_TAB_METADATA_TTL_MS = TABLE_METADATA_CACHE_TTL_MS;
|
||||
|
||||
function visibleQuerySortColumns(columns: string[], hiddenColumnIndexes: number[] | undefined, columnIndex: number): { resultColumns: string[]; columnIndex: number } | undefined {
|
||||
const hiddenIndexes = new Set(hiddenColumnIndexes ?? []);
|
||||
|
|
@ -84,8 +85,19 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
|
|||
const config = connectionStore.getConfig(target.connectionId);
|
||||
const querySchema = metadataSchemaForConnection(config, target.database, target.schema);
|
||||
console.info("[DBX][reloadData:metadata:get-columns:start]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), schema: querySchema, table: target.tableName });
|
||||
const columns = await api.getColumns(target.connectionId, target.database, querySchema, target.tableName, target.catalog);
|
||||
const indexes = await api.listIndexes(target.connectionId, target.database, querySchema, target.tableName, target.catalog).catch(() => []);
|
||||
// 复用共享表元数据缓存（30s TTL + in-flight 去重），多个入口对同一张表
|
||||
// 不再各自往返 getColumns/listIndexes
|
||||
const { metadata } = await loadTableMetadata({
|
||||
connectionId: target.connectionId,
|
||||
database: target.database,
|
||||
schema: querySchema,
|
||||
tableName: target.tableName,
|
||||
tableType: target.tableType,
|
||||
databaseType: effectiveDatabaseTypeForConnection(config) ?? config?.db_type ?? "",
|
||||
driverProfile: config?.driver_profile || config?.db_type,
|
||||
catalog: target.catalog,
|
||||
});
|
||||
const columns = metadata.columns;
|
||||
console.info("[DBX][reloadData:metadata:get-columns:done]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), columnCount: columns.length });
|
||||
const current = queryStore.tabs.find((item) => item.id === target.tabId);
|
||||
const currentMeta = current ? tableMetaForDataTab(current) : undefined;
|
||||
|
|
@ -93,7 +105,7 @@ export function useDataGridActions(activeTab: ComputedRef<QueryTab | undefined>)
|
|||
console.info("[DBX][reloadData:metadata:stale-tab]", { traceId: trace?.traceId, elapsed: trace?.elapsed(), table: target.tableName });
|
||||
return;
|
||||
}
|
||||
const primaryKeys = editableRowIdentifierColumns(effectiveDatabaseTypeForConnection(config), columns, indexes, target.tableType);
|
||||
const primaryKeys = metadata.primaryKeys;
|
||||
queryStore.setTableMeta(target.tabId, {
|
||||
catalog: target.catalog,
|
||||
schema: target.schema,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import * as api from "@/lib/backend/api";
|
||||
import { effectiveDatabaseTypeForConnection, metadataSchemaForConnection } from "@/lib/database/jdbcDialect";
|
||||
import { invalidateTableMetadataCache, loadTableMetadata } from "@/lib/metadata/tableMetadataCache";
|
||||
import { isNoSnapshotErrorResult } from "@/lib/query/queryResultError";
|
||||
import { buildTableSelectSql } from "@/lib/table/tableSelectSql";
|
||||
import { editableRowIdentifierColumns, usesSyntheticRowIdKey } from "@/lib/table/tableEditing";
|
||||
|
|
@ -7,7 +8,7 @@ import { tableOpenPageLimit } from "@/lib/table/tableOpenPageLimit";
|
|||
import { useConnectionStore } from "@/stores/connectionStore";
|
||||
import { useQueryStore } from "@/stores/queryStore";
|
||||
import { useSettingsStore } from "@/stores/settingsStore";
|
||||
import type { TableInfoTab } from "@/types/database";
|
||||
import type { ColumnInfo, TableInfoTab } from "@/types/database";
|
||||
|
||||
export type NavigationTarget = {
|
||||
connectionId: string;
|
||||
|
|
@ -145,9 +146,19 @@ async function openTableTarget(target: NavigationTarget, options: { tableInfoTab
|
|||
await queryStore.executeTabSql(tabId, emptySql, { pagination: { limit: pageLimit, offset: 0 } });
|
||||
}
|
||||
try {
|
||||
const columns = await api.getColumns(target.connectionId, target.database, querySchema, target.tableName, target.catalog);
|
||||
const indexes = await api.listIndexes(target.connectionId, target.database, querySchema, target.tableName, target.catalog).catch(() => []);
|
||||
const primaryKeys = editableRowIdentifierColumns(effectiveDbType, columns, indexes, targetTableType);
|
||||
// 复用共享表元数据缓存（30s TTL + in-flight 去重）
|
||||
const { metadata } = await loadTableMetadata({
|
||||
connectionId: target.connectionId,
|
||||
database: target.database,
|
||||
schema: querySchema,
|
||||
tableName: target.tableName,
|
||||
tableType: targetTableType,
|
||||
databaseType: effectiveDbType ?? config.db_type,
|
||||
driverProfile: config.driver_profile || config.db_type,
|
||||
catalog: target.catalog,
|
||||
});
|
||||
const columns = metadata.columns;
|
||||
const primaryKeys = metadata.primaryKeys;
|
||||
const useRowId = usesSyntheticRowIdKey(effectiveDbType, primaryKeys, targetTableType);
|
||||
queryStore.setTableMeta(tabId, {
|
||||
schema: target.schema,
|
||||
|
|
@ -216,17 +227,40 @@ export function useNavigationTargets(dialogs: { showFieldLineageDialog: { value:
|
|||
} catch {}
|
||||
}
|
||||
queryStore.invalidateTableStructure(context.connectionId, context.database, context.schema, context.tableName);
|
||||
// 结构已变更：无论是否有打开的 data tab 都必须作废共享元数据缓存，否则
|
||||
// 其它 loadTableMetadata 消费者最长 30 秒拿到旧列。不带 schema/catalog
|
||||
// 维度（宁可多废，schema 形态在各消费点可能不同）
|
||||
invalidateTableMetadataCache({ connectionId: context.connectionId, database: context.database, tableName: context.tableName });
|
||||
const matchingDataTabs = queryStore.tabs.filter((tab) => tab.mode === "data" && tab.connectionId === context.connectionId && tab.database === context.database && tab.tableMeta?.tableName === context.tableName && (tab.tableMeta.schema || "") === (context.schema || ""));
|
||||
// 同一 catalog 只强制加载一次，结果分发给全部匹配 tab
|
||||
const loadedByCatalog = new Map<string, { columns: ColumnInfo[]; primaryKeys: string[] }>();
|
||||
for (const tab of matchingDataTabs) {
|
||||
try {
|
||||
const connection = connectionStore.getConfig(tab.connectionId);
|
||||
const metadataSchema = metadataSchemaForConnection(connection, tab.database, tab.tableMeta?.schema);
|
||||
const columns = await api.getColumns(tab.connectionId, tab.database, metadataSchema, tab.tableMeta!.tableName);
|
||||
const indexes = await api.listIndexes(tab.connectionId, tab.database, metadataSchema, tab.tableMeta!.tableName).catch(() => []);
|
||||
// 分组含 tableType：主键计算依赖它，不同 tableType 不能共享加载结果
|
||||
const catalogKey = `${tab.tableMeta?.catalog ?? ""}\u0000${tab.tableMeta?.tableType ?? ""}`;
|
||||
let metadata = loadedByCatalog.get(catalogKey);
|
||||
if (!metadata) {
|
||||
metadata = (
|
||||
await loadTableMetadata({
|
||||
connectionId: tab.connectionId,
|
||||
database: tab.database,
|
||||
schema: metadataSchema,
|
||||
tableName: tab.tableMeta!.tableName,
|
||||
tableType: tab.tableMeta!.tableType,
|
||||
databaseType: effectiveDatabaseTypeForConnection(connection) ?? connection?.db_type ?? "",
|
||||
driverProfile: connection?.driver_profile || connection?.db_type,
|
||||
catalog: tab.tableMeta?.catalog,
|
||||
force: true,
|
||||
})
|
||||
).metadata;
|
||||
loadedByCatalog.set(catalogKey, metadata);
|
||||
}
|
||||
queryStore.setTableMeta(tab.id, {
|
||||
...tab.tableMeta!,
|
||||
columns,
|
||||
primaryKeys: editableRowIdentifierColumns(effectiveDatabaseTypeForConnection(connection), columns, indexes, tab.tableMeta!.tableType),
|
||||
columns: metadata.columns,
|
||||
primaryKeys: metadata.primaryKeys,
|
||||
});
|
||||
if (tab.id === queryStore.activeTabId) await reloadData();
|
||||
} catch (e: any) {
|
||||
|
|
|
|||
119
apps/desktop/src/lib/metadata/__tests__/tableMetadataCache.invalidation.spec.ts
vendored
Normal file
119
apps/desktop/src/lib/metadata/__tests__/tableMetadataCache.invalidation.spec.ts
vendored
Normal file
|
|
@ -0,0 +1,119 @@
|
|||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { ColumnInfo } from "@/types/database";
|
||||
|
||||
const mocks = vi.hoisted(() => ({
|
||||
getColumns: vi.fn(),
|
||||
listIndexes: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/backend/api", () => ({
|
||||
getColumns: mocks.getColumns,
|
||||
listIndexes: mocks.listIndexes,
|
||||
}));
|
||||
|
||||
import { clearTableMetadataCache, getCachedTableMetadata, invalidateTableMetadataCache, loadTableMetadata } from "@/lib/metadata/tableMetadataCache";
|
||||
|
||||
function column(name: string): ColumnInfo {
|
||||
return { name, data_type: "integer", is_nullable: false, column_default: null, is_primary_key: true, extra: null };
|
||||
}
|
||||
|
||||
const request = { connectionId: "c1", database: "db", schema: "public", tableName: "users", databaseType: "postgres" } as const;
|
||||
|
||||
describe("tableMetadataCache invalidation", () => {
|
||||
beforeEach(() => {
|
||||
clearTableMetadataCache();
|
||||
vi.clearAllMocks();
|
||||
mocks.listIndexes.mockResolvedValue([]);
|
||||
});
|
||||
|
||||
it("a stale in-flight load cannot overwrite the cache after invalidation", async () => {
|
||||
// 旧加载（结构变更前启动）被挂起
|
||||
let releaseOldColumns: (columns: ColumnInfo[]) => void = () => {};
|
||||
mocks.getColumns.mockReturnValueOnce(
|
||||
new Promise<ColumnInfo[]>((resolve) => {
|
||||
releaseOldColumns = resolve;
|
||||
}),
|
||||
);
|
||||
const oldLoad = loadTableMetadata({ ...request });
|
||||
|
||||
// 结构保存：作废缓存并 force 拉新
|
||||
invalidateTableMetadataCache({ connectionId: "c1", database: "db", tableName: "users" });
|
||||
mocks.getColumns.mockResolvedValueOnce([column("id_new")]);
|
||||
const fresh = await loadTableMetadata({ ...request, force: true });
|
||||
expect(fresh.metadata.columns[0]?.name).toBe("id_new");
|
||||
|
||||
// 旧加载最后返回：不得回填覆盖新值
|
||||
releaseOldColumns([column("id_old")]);
|
||||
const oldResult = await oldLoad;
|
||||
expect(oldResult.metadata.columns[0]?.name).toBe("id_old");
|
||||
expect(getCachedTableMetadata(request)?.metadata.columns[0]?.name).toBe("id_new");
|
||||
});
|
||||
|
||||
it("a follower starting after invalidation does not join the stale in-flight load", async () => {
|
||||
// 旧加载（失效前启动、non-force、已登记在途）被挂起
|
||||
let releaseOldColumns: (columns: ColumnInfo[]) => void = () => {};
|
||||
mocks.getColumns.mockReturnValueOnce(
|
||||
new Promise<ColumnInfo[]>((resolve) => {
|
||||
releaseOldColumns = resolve;
|
||||
}),
|
||||
);
|
||||
const oldLoad = loadTableMetadata({ ...request });
|
||||
|
||||
invalidateTableMetadataCache({ connectionId: "c1", database: "db", tableName: "users" });
|
||||
|
||||
// 失效后的 non-force 跟随者：不得加入旧在途（否则其失效代数取自失效后，
|
||||
// 完成时会把旧结果写回缓存），必须自起新加载
|
||||
mocks.getColumns.mockResolvedValueOnce([column("id_new")]);
|
||||
const follower = await loadTableMetadata({ ...request });
|
||||
expect(follower.metadata.columns[0]?.name).toBe("id_new");
|
||||
expect(mocks.getColumns).toHaveBeenCalledTimes(2);
|
||||
|
||||
releaseOldColumns([column("id_old")]);
|
||||
await oldLoad;
|
||||
expect(getCachedTableMetadata(request)?.metadata.columns[0]?.name).toBe("id_new");
|
||||
});
|
||||
|
||||
it("invalidating one table does not disturb another table's in-flight load", async () => {
|
||||
const requestB = { ...request, tableName: "orders" } as const;
|
||||
// A、B 同时挂起
|
||||
const releases = new Map<string, (columns: ColumnInfo[]) => void>();
|
||||
mocks.getColumns.mockImplementation(
|
||||
(_connectionId: string, _database: string, _schema: string, tableName: string) =>
|
||||
new Promise<ColumnInfo[]>((resolve) => {
|
||||
releases.set(tableName, resolve);
|
||||
}),
|
||||
);
|
||||
const loadA = loadTableMetadata({ ...request });
|
||||
const loadB = loadTableMetadata({ ...requestB });
|
||||
await Promise.resolve();
|
||||
|
||||
// 仅失效 A
|
||||
invalidateTableMetadataCache({ connectionId: "c1", database: "db", tableName: "users" });
|
||||
|
||||
// B 的 follower 必须复用原在途请求：总请求数保持 2（A 一次 + B 一次）。
|
||||
// 先让出 microtask，coordinator 的 load 才会真正启动，断言才有观察力
|
||||
const followerB = loadTableMetadata({ ...requestB });
|
||||
await Promise.resolve();
|
||||
expect(mocks.getColumns).toHaveBeenCalledTimes(2);
|
||||
|
||||
// B 返回后正常写入缓存
|
||||
releases.get("orders")?.([column("order_id")]);
|
||||
expect((await loadB).metadata.columns[0]?.name).toBe("order_id");
|
||||
expect((await followerB).metadata.columns[0]?.name).toBe("order_id");
|
||||
expect(getCachedTableMetadata(requestB)?.metadata.columns[0]?.name).toBe("order_id");
|
||||
|
||||
// A 跨越失效边界：完成后不得写缓存
|
||||
releases.get("users")?.([column("id_old")]);
|
||||
await loadA;
|
||||
expect(getCachedTableMetadata(request)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("caches normally when no invalidation crosses the load", async () => {
|
||||
mocks.getColumns.mockResolvedValueOnce([column("id")]);
|
||||
await loadTableMetadata({ ...request });
|
||||
expect(getCachedTableMetadata(request)?.metadata.columns[0]?.name).toBe("id");
|
||||
// 命中缓存：不再发起请求
|
||||
await loadTableMetadata({ ...request });
|
||||
expect(mocks.getColumns).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
|
@ -40,6 +40,12 @@ function valuesEqual(left: unknown, right: unknown): boolean {
|
|||
return left === right;
|
||||
}
|
||||
|
||||
/** 将失效 match 编译为作用于 metadataScopeParts 结果的谓词，供缓存外的在途登记复用同一匹配语义 */
|
||||
export function metadataCacheInvalidationMatcher(match: MetadataCacheInvalidation): (scope: ReturnType<typeof metadataScopeParts>) => boolean {
|
||||
const normalized = normalizeInvalidation(match);
|
||||
return (scope) => Object.entries(normalized).every(([field, value]) => valuesEqual(scope[field as keyof typeof scope], value));
|
||||
}
|
||||
|
||||
export class MetadataResultCache<T> {
|
||||
private readonly entries = new Map<string, MetadataCacheEntry<T>>();
|
||||
private readonly now: () => number;
|
||||
|
|
@ -69,11 +75,10 @@ export class MetadataResultCache<T> {
|
|||
}
|
||||
|
||||
invalidate(match: MetadataCacheInvalidation): number {
|
||||
const normalized = normalizeInvalidation(match);
|
||||
const matches = metadataCacheInvalidationMatcher(match);
|
||||
let removed = 0;
|
||||
for (const [key, entry] of this.entries) {
|
||||
const matches = Object.entries(normalized).every(([field, value]) => valuesEqual(entry.scope[field as keyof typeof entry.scope], value));
|
||||
if (matches) {
|
||||
if (matches(entry.scope)) {
|
||||
this.entries.delete(key);
|
||||
removed++;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@ import type { ColumnInfo, DatabaseType, IndexInfo, QueryTab } from "@/types/data
|
|||
import * as api from "@/lib/backend/api";
|
||||
import { editableRowIdentifierColumns } from "@/lib/table/tableEditing";
|
||||
import { createMetadataLoadTrace, logMetadataLoadTrace, MetadataLoadCoordinator, type MetadataLoadCacheStatus, type MetadataLoadTraceLogger } from "./metadataLoadCoordinator";
|
||||
import type { MetadataScopeInput } from "./metadataLoadScope";
|
||||
import { MetadataResultCache, type MetadataCacheInvalidation } from "./metadataResultCache";
|
||||
import { metadataScopeKey, metadataScopeParts, type MetadataScopeInput } from "./metadataLoadScope";
|
||||
import { metadataCacheInvalidationMatcher, MetadataResultCache, type MetadataCacheInvalidation } from "./metadataResultCache";
|
||||
|
||||
export const TABLE_METADATA_CACHE_TTL_MS = 30_000;
|
||||
const TABLE_METADATA_CACHE_MAX_ENTRIES = 120;
|
||||
|
|
@ -48,6 +48,41 @@ const tableMetadataCoordinator = new MetadataLoadCoordinator((event) => {
|
|||
console.debug("[DBX][metadata-load:table-coordinator]", event);
|
||||
});
|
||||
|
||||
// 失效代数（按 scope key 隔离）：跨越失效边界的旧加载完成后不得写缓存——
|
||||
// 结构变更后 force 拉到的新值可能被保存前启动、最后返回的在途加载回填覆盖。
|
||||
// 只登记在途 scope 并只对匹配失效条件的 key 递增代数，避免失效表 A 时
|
||||
// 波及无关表 B 的在途去重与缓存写入
|
||||
interface InFlightTableMetadataScope {
|
||||
parts: ReturnType<typeof metadataScopeParts>;
|
||||
count: number;
|
||||
}
|
||||
const inFlightTableMetadataScopes = new Map<string, InFlightTableMetadataScope>();
|
||||
const tableMetadataInvalidationStamps = new Map<string, number>();
|
||||
|
||||
function registerInFlightTableMetadataScope(scopeKey: string, scope: MetadataScopeInput): void {
|
||||
const entry = inFlightTableMetadataScopes.get(scopeKey);
|
||||
if (entry) {
|
||||
entry.count++;
|
||||
} else {
|
||||
inFlightTableMetadataScopes.set(scopeKey, { parts: metadataScopeParts(scope), count: 1 });
|
||||
}
|
||||
}
|
||||
|
||||
function unregisterInFlightTableMetadataScope(scopeKey: string): void {
|
||||
const entry = inFlightTableMetadataScopes.get(scopeKey);
|
||||
if (!entry) return;
|
||||
entry.count--;
|
||||
if (entry.count > 0) return;
|
||||
inFlightTableMetadataScopes.delete(scopeKey);
|
||||
// 代数只在加载的 start→end 窗口内比较，且失效只 bump 在途 key；
|
||||
// 无在途加载时清掉代数，两张 Map 都随在途集合有界
|
||||
tableMetadataInvalidationStamps.delete(scopeKey);
|
||||
}
|
||||
|
||||
function bumpTableMetadataInvalidationStamp(scopeKey: string): void {
|
||||
tableMetadataInvalidationStamps.set(scopeKey, (tableMetadataInvalidationStamps.get(scopeKey) ?? 0) + 1);
|
||||
}
|
||||
|
||||
export function tableMetadataScope(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog">): MetadataScopeInput {
|
||||
return {
|
||||
kind: "table-metadata",
|
||||
|
|
@ -95,28 +130,39 @@ export async function loadTableMetadata(request: TableMetadataRequest): Promise<
|
|||
}
|
||||
|
||||
logMetadataLoadTrace(request.traceLogger, trace, "cache-miss", { cacheStatus: request.force ? "refresh" : "miss", force: request.force === true });
|
||||
const metadata = await tableMetadataCoordinator.run(
|
||||
scope,
|
||||
async () => {
|
||||
const columns = await api.getColumns(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog);
|
||||
const indexes = await api.listIndexes(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog).catch((): IndexInfo[] => []);
|
||||
const primaryKeys = editableRowIdentifierColumns(request.databaseType as DatabaseType, columns, indexes, request.tableType);
|
||||
return {
|
||||
schema: request.schema || undefined,
|
||||
tableName: request.tableName,
|
||||
tableType: request.tableType,
|
||||
catalog: request.catalog,
|
||||
database: request.database,
|
||||
columns,
|
||||
indexes,
|
||||
primaryKeys,
|
||||
cachedAt: Date.now(),
|
||||
};
|
||||
},
|
||||
{ force: request.force, kind: scope.kind },
|
||||
);
|
||||
const scopeKey = metadataScopeKey(scope);
|
||||
const invalidationStampAtStart = tableMetadataInvalidationStamps.get(scopeKey) ?? 0;
|
||||
registerInFlightTableMetadataScope(scopeKey, scope);
|
||||
let metadata: TableMetadata;
|
||||
try {
|
||||
metadata = await tableMetadataCoordinator.run(
|
||||
scope,
|
||||
async () => {
|
||||
const columns = await api.getColumns(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog);
|
||||
const indexes = await api.listIndexes(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog).catch((): IndexInfo[] => []);
|
||||
const primaryKeys = editableRowIdentifierColumns(request.databaseType as DatabaseType, columns, indexes, request.tableType);
|
||||
return {
|
||||
schema: request.schema || undefined,
|
||||
tableName: request.tableName,
|
||||
tableType: request.tableType,
|
||||
catalog: request.catalog,
|
||||
database: request.database,
|
||||
columns,
|
||||
indexes,
|
||||
primaryKeys,
|
||||
cachedAt: Date.now(),
|
||||
};
|
||||
},
|
||||
{ force: request.force, kind: scope.kind },
|
||||
);
|
||||
|
||||
tableMetadataCache.set(scope, metadata);
|
||||
// 必须在 unregister 前比较：最后一个在途加载注销时会顺带清掉代数记录
|
||||
if (invalidationStampAtStart === (tableMetadataInvalidationStamps.get(scopeKey) ?? 0)) {
|
||||
tableMetadataCache.set(scope, metadata);
|
||||
}
|
||||
} finally {
|
||||
unregisterInFlightTableMetadataScope(scopeKey);
|
||||
}
|
||||
logMetadataLoadTrace(request.traceLogger, trace, "done", {
|
||||
cacheStatus: request.force ? "refresh" : "miss",
|
||||
resultCount: metadata.columns.length,
|
||||
|
|
@ -126,9 +172,23 @@ export async function loadTableMetadata(request: TableMetadataRequest): Promise<
|
|||
}
|
||||
|
||||
export function invalidateTableMetadataCache(match: MetadataCacheInvalidation): number {
|
||||
// 只处理匹配失效条件的在途 scope，不波及其他表/连接的在途去重与缓存写入：
|
||||
// 1) bump 该 scope 的失效代数——跨边界的旧加载完成后不得写缓存；
|
||||
// 2) 甩掉该 scope 的在途登记——否则失效后启动的 non-force 调用会加入
|
||||
// 失效前的旧加载，且其失效代数取自失效之后，完成时把旧结果写回缓存
|
||||
const matches = metadataCacheInvalidationMatcher(match);
|
||||
for (const [scopeKey, entry] of inFlightTableMetadataScopes) {
|
||||
if (!matches(entry.parts)) continue;
|
||||
bumpTableMetadataInvalidationStamp(scopeKey);
|
||||
tableMetadataCoordinator.clear(scopeKey);
|
||||
}
|
||||
return tableMetadataCache.invalidate(match);
|
||||
}
|
||||
|
||||
export function clearTableMetadataCache(): void {
|
||||
for (const scopeKey of inFlightTableMetadataScopes.keys()) {
|
||||
bumpTableMetadataInvalidationStamp(scopeKey);
|
||||
}
|
||||
tableMetadataCoordinator.clear();
|
||||
tableMetadataCache.clear();
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue