From af9fd77780eadff8ac28145a73fc213f13db2fff Mon Sep 17 00:00:00 2001 From: Andrey Avtomonov Date: Wed, 13 May 2026 00:01:15 +0200 Subject: [PATCH] fix(context): expose warehouse targets for LookML and MetricFlow --- .../adapters/lookml/lookml.adapter.test.ts | 12 ++++++++ .../ingest/adapters/lookml/lookml.adapter.ts | 9 ++++++ .../metricflow/metricflow.adapter.test.ts | 9 ++++++ .../adapters/metricflow/metricflow.adapter.ts | 9 ++++++ .../context/src/ingest/local-adapters.test.ts | 29 +++++++++++++++++++ packages/context/src/ingest/local-adapters.ts | 10 +++++-- 6 files changed, 76 insertions(+), 2 deletions(-) diff --git a/packages/context/src/ingest/adapters/lookml/lookml.adapter.test.ts b/packages/context/src/ingest/adapters/lookml/lookml.adapter.test.ts index 32564da1..d22597b9 100644 --- a/packages/context/src/ingest/adapters/lookml/lookml.adapter.test.ts +++ b/packages/context/src/ingest/adapters/lookml/lookml.adapter.test.ts @@ -15,6 +15,18 @@ describe('LookmlSourceAdapter validation sidecars', () => { afterEach(async () => rm(tmpRoot, { recursive: true, force: true })); + it('returns configured target warehouse connection ids', async () => { + const adapter = new LookmlSourceAdapter({ + homeDir: join(tmpRoot, 'home'), + targetConnectionIds: ['warehouse', 'analytics', 'warehouse'], + }); + + await expect(adapter.listTargetConnectionIds?.(join(tmpRoot, 'staged'))).resolves.toEqual([ + 'analytics', + 'warehouse', + ]); + }); + it('writes a partial fetch report and marks mismatched chunks as SL-disallowed', async () => { const originRoot = join(tmpRoot, 'origin-src'); await mkdir(join(originRoot, 'views'), { recursive: true }); diff --git a/packages/context/src/ingest/adapters/lookml/lookml.adapter.ts b/packages/context/src/ingest/adapters/lookml/lookml.adapter.ts index 9978ddd4..3ea7b9a6 100644 --- a/packages/context/src/ingest/adapters/lookml/lookml.adapter.ts +++ b/packages/context/src/ingest/adapters/lookml/lookml.adapter.ts @@ -14,6 +14,11 @@ import { parseLookmlPullConfig } from './pull-config.js'; export interface LookmlSourceAdapterDeps { homeDir: string; + targetConnectionIds?: string[]; +} + +function uniqueSorted(values: readonly string[] | undefined): string[] { + return [...new Set(values ?? [])].sort((left, right) => left.localeCompare(right)); } export class LookmlSourceAdapter implements SourceAdapter { @@ -43,6 +48,10 @@ export class LookmlSourceAdapter implements SourceAdapter { return readLookmlFetchReport(stagedDir); } + async listTargetConnectionIds(_stagedDir: string): Promise { + return uniqueSorted(this.deps.targetConnectionIds); + } + async chunk(stagedDir: string, diffSet?: DiffSet): Promise { const project = await parseLookmlStagedDir(stagedDir); const mismatchedModelNames = await readLookmlMismatchedModelNames(stagedDir); diff --git a/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.test.ts b/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.test.ts index af2e409f..19bb6cdc 100644 --- a/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.test.ts +++ b/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.test.ts @@ -42,6 +42,15 @@ describe('MetricflowSourceAdapter', () => { expect(adapter.skillNames).toEqual(['metricflow_ingest']); }); + it('returns configured target warehouse connection ids', async () => { + const metricflow = new MetricflowSourceAdapter({ + homeDir: join(tmpRoot, 'cache-home'), + targetConnectionIds: ['warehouse', 'analytics', 'warehouse'], + }); + + await expect(metricflow.listTargetConnectionIds?.(stagedDir)).resolves.toEqual(['analytics', 'warehouse']); + }); + it('detects a staged dir with a semantic_models YAML', async () => { await mkdir(join(stagedDir, 'models'), { recursive: true }); await writeFile( diff --git a/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.ts b/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.ts index a465a973..c8182ed8 100644 --- a/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.ts +++ b/packages/context/src/ingest/adapters/metricflow/metricflow.adapter.ts @@ -9,6 +9,11 @@ import { parseMetricflowPullConfig } from './pull-config.js'; export interface MetricflowSourceAdapterDeps { homeDir: string; + targetConnectionIds?: string[]; +} + +function uniqueSorted(values: readonly string[] | undefined): string[] { + return [...new Set(values ?? [])].sort((left, right) => left.localeCompare(right)); } export class MetricflowSourceAdapter implements SourceAdapter { @@ -30,6 +35,10 @@ export class MetricflowSourceAdapter implements SourceAdapter { }); } + async listTargetConnectionIds(_stagedDir: string): Promise { + return uniqueSorted(this.deps.targetConnectionIds); + } + async chunk(stagedDir: string, diffSet?: DiffSet): Promise { const project = await parseMetricFlowStagedDir(stagedDir); const chunk = await chunkMetricFlowProject(project, { diffSet }); diff --git a/packages/context/src/ingest/local-adapters.test.ts b/packages/context/src/ingest/local-adapters.test.ts index 9fd51d01..7161743a 100644 --- a/packages/context/src/ingest/local-adapters.test.ts +++ b/packages/context/src/ingest/local-adapters.test.ts @@ -523,6 +523,35 @@ describe('local ingest adapters', () => { await expect(notion?.listTargetConnectionIds?.('/tmp/staged-notion')).resolves.toEqual(['warehouse']); }); + it('passes primary warehouse connection ids to local LookML and MetricFlow adapters', async () => { + const adapters = createDefaultLocalIngestAdapters( + projectWithConnections({ + warehouse: { + driver: 'postgres', + url: 'postgresql://readonly@db.example.test/analytics', + }, + lookml_docs: { + driver: 'lookml', + lookml: { + repoUrl: 'https://github.com/acme/lookml.git', + }, + }, + metrics_repo: { + driver: 'metricflow', + metricflow: { + repoUrl: 'https://github.com/acme/metrics.git', + }, + }, + } as never), + ); + + const lookml = adapters.find((adapter) => adapter.source === 'lookml'); + const metricflow = adapters.find((adapter) => adapter.source === 'metricflow'); + + await expect(lookml?.listTargetConnectionIds?.('/tmp/staged-lookml')).resolves.toEqual(['warehouse']); + await expect(metricflow?.listTargetConnectionIds?.('/tmp/staged-metricflow')).resolves.toEqual(['warehouse']); + }); + it('resolves MetricFlow auth_token_ref without writing literal tokens to config', async () => { const project = projectWithConnections({ metricflow_main: { diff --git a/packages/context/src/ingest/local-adapters.ts b/packages/context/src/ingest/local-adapters.ts index 14c6b683..533bd526 100644 --- a/packages/context/src/ingest/local-adapters.ts +++ b/packages/context/src/ingest/local-adapters.ts @@ -88,7 +88,10 @@ export function createDefaultLocalIngestAdapters( ...(options.databaseIntrospectionUrl ? { baseUrl: options.databaseIntrospectionUrl } : {}), }), }), - new LookmlSourceAdapter({ homeDir: join(project.projectDir, '.ktx/cache') }), + new LookmlSourceAdapter({ + homeDir: join(project.projectDir, '.ktx/cache'), + targetConnectionIds: primaryWarehouseConnectionIds(project), + }), new DbtSourceAdapter({ homeDir: join(project.projectDir, '.ktx/cache'), targetConnectionIds: primaryWarehouseConnectionIds(project), @@ -106,7 +109,10 @@ export function createDefaultLocalIngestAdapters( }, }, }), - new MetricflowSourceAdapter({ homeDir: join(project.projectDir, '.ktx/cache') }), + new MetricflowSourceAdapter({ + homeDir: join(project.projectDir, '.ktx/cache'), + targetConnectionIds: primaryWarehouseConnectionIds(project), + }), new NotionSourceAdapter({ targetConnectionIds: primaryWarehouseConnectionIds(project), ...(options.logger ? { logger: options.logger } : {}),