mirror of
https://github.com/Kaelio/ktx.git
synced 2026-07-07 11:02:11 +02:00
feat(athena): first-class AWS Athena warehouse identity (SL + BI mapping) (#332)
* feat: support Athena warehouse identity * fix: support Athena and DuckDB BI table parsing
This commit is contained in:
parent
fe7e6bd1fa
commit
f310391da5
14 changed files with 252 additions and 45 deletions
|
|
@ -0,0 +1,24 @@
|
||||||
|
import type { ConnectionType } from './connection-type.js';
|
||||||
|
|
||||||
|
const CONNECTION_TYPE_TO_SQLGLOT = {
|
||||||
|
POSTGRESQL: 'postgres',
|
||||||
|
SQLITE: 'sqlite',
|
||||||
|
DUCKDB: 'duckdb',
|
||||||
|
SQLSERVER: 'tsql',
|
||||||
|
BIGQUERY: 'bigquery',
|
||||||
|
SNOWFLAKE: 'snowflake',
|
||||||
|
MYSQL: 'mysql',
|
||||||
|
CLICKHOUSE: 'clickhouse',
|
||||||
|
ATHENA: 'athena',
|
||||||
|
METABASE: null,
|
||||||
|
LOOKER: null,
|
||||||
|
NOTION: null,
|
||||||
|
} satisfies Record<ConnectionType, string | null>;
|
||||||
|
|
||||||
|
export function dialectForConnectionType(connectionType: string): string {
|
||||||
|
return CONNECTION_TYPE_TO_SQLGLOT[connectionType.toUpperCase() as ConnectionType] ?? 'postgres';
|
||||||
|
}
|
||||||
|
|
||||||
|
export function warehouseTargetDialect(connectionType: string): string | null {
|
||||||
|
return CONNECTION_TYPE_TO_SQLGLOT[connectionType.toUpperCase() as ConnectionType] ?? null;
|
||||||
|
}
|
||||||
|
|
@ -7,22 +7,12 @@ export const connectionTypeSchema = z.enum([
|
||||||
'SQLSERVER',
|
'SQLSERVER',
|
||||||
'BIGQUERY',
|
'BIGQUERY',
|
||||||
'SNOWFLAKE',
|
'SNOWFLAKE',
|
||||||
'CENTRALREACH',
|
|
||||||
'EPIC',
|
|
||||||
'CERNER',
|
|
||||||
'ATHENA',
|
'ATHENA',
|
||||||
'QUICKBOOKS',
|
|
||||||
'WORKDAY',
|
|
||||||
'REST',
|
|
||||||
'S3',
|
|
||||||
'SLACK',
|
|
||||||
'METABASE',
|
'METABASE',
|
||||||
'LOOKER',
|
'LOOKER',
|
||||||
'NOTION',
|
'NOTION',
|
||||||
'MYSQL',
|
'MYSQL',
|
||||||
'CLICKHOUSE',
|
'CLICKHOUSE',
|
||||||
'PLAIN',
|
|
||||||
'BETTERSTACK',
|
|
||||||
]);
|
]);
|
||||||
|
|
||||||
export type ConnectionType = z.infer<typeof connectionTypeSchema>;
|
export type ConnectionType = z.infer<typeof connectionTypeSchema>;
|
||||||
|
|
|
||||||
|
|
@ -29,6 +29,7 @@ const DRIVER_TO_CONNECTION_TYPE: Record<string, ConnectionType> = {
|
||||||
clickhouse: 'CLICKHOUSE',
|
clickhouse: 'CLICKHOUSE',
|
||||||
snowflake: 'SNOWFLAKE',
|
snowflake: 'SNOWFLAKE',
|
||||||
bigquery: 'BIGQUERY',
|
bigquery: 'BIGQUERY',
|
||||||
|
athena: 'ATHENA',
|
||||||
};
|
};
|
||||||
|
|
||||||
export function localConnectionToWarehouseDescriptor(
|
export function localConnectionToWarehouseDescriptor(
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
import type { ParsedTargetTable } from '../../parsed-target-table.js';
|
import type { ParsedTargetTable } from '../../parsed-target-table.js';
|
||||||
|
import { warehouseTargetDialect } from '../../../connections/connection-type-dialect.js';
|
||||||
import type { LookerWarehouseConnectionInfo } from './client.js';
|
import type { LookerWarehouseConnectionInfo } from './client.js';
|
||||||
import type { LookerPullConfig, LookerRuntimeCursors, StagedExploreFile, StagedLookmlModelsFile } from './types.js';
|
import type { LookerPullConfig, LookerRuntimeCursors, StagedExploreFile, StagedLookmlModelsFile } from './types.js';
|
||||||
|
|
||||||
|
|
@ -11,6 +12,7 @@ const LOOKER_DIALECT_TO_CONNECTION_TYPE = {
|
||||||
sqlite: 'SQLITE',
|
sqlite: 'SQLITE',
|
||||||
sqlserver: 'SQLSERVER',
|
sqlserver: 'SQLSERVER',
|
||||||
clickhouse: 'CLICKHOUSE',
|
clickhouse: 'CLICKHOUSE',
|
||||||
|
awsathena: 'ATHENA',
|
||||||
} as const;
|
} as const;
|
||||||
|
|
||||||
/** @internal */
|
/** @internal */
|
||||||
|
|
@ -72,16 +74,6 @@ export interface LookerMappingClient {
|
||||||
getExplore(modelName: string, exploreName: string): Promise<StagedExploreFile>;
|
getExplore(modelName: string, exploreName: string): Promise<StagedExploreFile>;
|
||||||
}
|
}
|
||||||
|
|
||||||
const SQLGLOT_DIALECT_BY_CONNECTION_TYPE: Partial<Record<LookerWarehouseTargetConnectionType, string>> = {
|
|
||||||
BIGQUERY: 'bigquery',
|
|
||||||
SNOWFLAKE: 'snowflake',
|
|
||||||
POSTGRESQL: 'postgres',
|
|
||||||
MYSQL: 'mysql',
|
|
||||||
SQLITE: 'sqlite',
|
|
||||||
SQLSERVER: 'tsql',
|
|
||||||
CLICKHOUSE: 'clickhouse',
|
|
||||||
};
|
|
||||||
|
|
||||||
export async function discoverLookerConnections(
|
export async function discoverLookerConnections(
|
||||||
client: Pick<LookerMappingClient, 'listLookerConnections'>,
|
client: Pick<LookerMappingClient, 'listLookerConnections'>,
|
||||||
): Promise<LookerWarehouseConnectionInfo[]> {
|
): Promise<LookerWarehouseConnectionInfo[]> {
|
||||||
|
|
@ -100,7 +92,7 @@ export function lookerDialectToConnectionType(dialect: string | null): LookerWar
|
||||||
|
|
||||||
/** @internal */
|
/** @internal */
|
||||||
export function sqlglotDialectForConnectionType(connectionType: string): string | null {
|
export function sqlglotDialectForConnectionType(connectionType: string): string | null {
|
||||||
return SQLGLOT_DIALECT_BY_CONNECTION_TYPE[connectionType as LookerWarehouseTargetConnectionType] ?? null;
|
return warehouseTargetDialect(connectionType);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** @internal */
|
/** @internal */
|
||||||
|
|
|
||||||
|
|
@ -8,9 +8,9 @@ export const METABASE_ENGINE_TO_CONNECTION_TYPE = {
|
||||||
snowflake: 'SNOWFLAKE',
|
snowflake: 'SNOWFLAKE',
|
||||||
sqlserver: 'SQLSERVER',
|
sqlserver: 'SQLSERVER',
|
||||||
mysql: 'MYSQL',
|
mysql: 'MYSQL',
|
||||||
|
athena: 'ATHENA',
|
||||||
} as const;
|
} as const;
|
||||||
|
|
||||||
|
|
||||||
export interface DiscoveredMetabaseDatabase {
|
export interface DiscoveredMetabaseDatabase {
|
||||||
id: number;
|
id: number;
|
||||||
name: string;
|
name: string;
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ import YAML from 'yaml';
|
||||||
import type { KtxFileStorePort } from '../../context/core/file-store.js';
|
import type { KtxFileStorePort } from '../../context/core/file-store.js';
|
||||||
import type { KtxLogger } from '../../context/core/config.js';
|
import type { KtxLogger } from '../../context/core/config.js';
|
||||||
import { noopLogger } from '../../context/core/config.js';
|
import { noopLogger } from '../../context/core/config.js';
|
||||||
|
import { dialectForConnectionType } from '../connections/connection-type-dialect.js';
|
||||||
import type { TableUsageOutput } from '../ingest/adapters/historic-sql/skill-schemas.js';
|
import type { TableUsageOutput } from '../ingest/adapters/historic-sql/skill-schemas.js';
|
||||||
import type { SlConnectionCatalogPort, SlPythonPort } from './ports.js';
|
import type { SlConnectionCatalogPort, SlPythonPort } from './ports.js';
|
||||||
import { normalizeSemanticLayerDescriptions } from './description-normalization.js';
|
import { normalizeSemanticLayerDescriptions } from './description-normalization.js';
|
||||||
|
|
@ -674,7 +675,7 @@ export class SemanticLayerService {
|
||||||
if (!connection) {
|
if (!connection) {
|
||||||
throw new Error(`Data source not found: ${connectionId}`);
|
throw new Error(`Data source not found: ${connectionId}`);
|
||||||
}
|
}
|
||||||
return SemanticLayerService.mapDialect(connection.connectionType);
|
return dialectForConnectionType(connection.connectionType);
|
||||||
}
|
}
|
||||||
|
|
||||||
async listFilesForConnection(connectionId: string): Promise<string[]> {
|
async listFilesForConnection(connectionId: string): Promise<string[]> {
|
||||||
|
|
@ -1107,25 +1108,6 @@ export class SemanticLayerService {
|
||||||
return { columns, tables };
|
return { columns, tables };
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* All callers should use this instead of maintaining their own dialect maps.
|
|
||||||
*/
|
|
||||||
static mapDialect(connectionType: string): string {
|
|
||||||
const normalized = connectionType.toUpperCase();
|
|
||||||
const map: Record<string, string> = {
|
|
||||||
POSTGRES: 'postgres',
|
|
||||||
BIGQUERY: 'bigquery',
|
|
||||||
SNOWFLAKE: 'snowflake',
|
|
||||||
MYSQL: 'mysql',
|
|
||||||
SQLSERVER: 'tsql',
|
|
||||||
SQLITE: 'sqlite',
|
|
||||||
DUCKDB: 'duckdb',
|
|
||||||
CLICKHOUSE: 'clickhouse',
|
|
||||||
DATABRICKS: 'databricks',
|
|
||||||
};
|
|
||||||
return map[normalized] ?? 'postgres';
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Execute a semantic layer query: load composed sources, generate SQL via
|
* Execute a semantic layer query: load composed sources, generate SQL via
|
||||||
* the python SL engine, and execute the generated SQL against the data source.
|
* the python SL engine, and execute the generated SQL against the data source.
|
||||||
|
|
@ -1153,7 +1135,7 @@ export class SemanticLayerService {
|
||||||
if (!connection) {
|
if (!connection) {
|
||||||
throw new Error(`Data source not found: ${connectionId}`);
|
throw new Error(`Data source not found: ${connectionId}`);
|
||||||
}
|
}
|
||||||
const dialect = SemanticLayerService.mapDialect(connection.connectionType);
|
const dialect = dialectForConnectionType(connection.connectionType);
|
||||||
|
|
||||||
// 3. Generate SQL via python SL engine
|
// 3. Generate SQL via python SL engine
|
||||||
const { data: slResult, error: slError } = await this.python.query({
|
const { data: slResult, error: slError } = await this.python.query({
|
||||||
|
|
|
||||||
|
|
@ -2,9 +2,10 @@ import YAML from 'yaml';
|
||||||
import type { GitService } from '../../../context/core/git.service.js';
|
import type { GitService } from '../../../context/core/git.service.js';
|
||||||
import type { KtxFileListResult, KtxFileReadResult, KtxFileStorePort } from '../../../context/core/file-store.js';
|
import type { KtxFileListResult, KtxFileReadResult, KtxFileStorePort } from '../../../context/core/file-store.js';
|
||||||
import { SYSTEM_GIT_AUTHOR } from '../../../context/tools/authors.js';
|
import { SYSTEM_GIT_AUTHOR } from '../../../context/tools/authors.js';
|
||||||
|
import { dialectForConnectionType } from '../../connections/connection-type-dialect.js';
|
||||||
import type { SlConnectionCatalogPort, SlSourcesIndexPort } from '../ports.js';
|
import type { SlConnectionCatalogPort, SlSourcesIndexPort } from '../ports.js';
|
||||||
import { sourceOverlaySchema } from '../schemas.js';
|
import { sourceOverlaySchema } from '../schemas.js';
|
||||||
import { SemanticLayerService } from '../semantic-layer.service.js';
|
import type { SemanticLayerService } from '../semantic-layer.service.js';
|
||||||
import { resolveSlSourceFile, slSourceFilePath } from '../source-files.js';
|
import { resolveSlSourceFile, slSourceFilePath } from '../source-files.js';
|
||||||
import type { SemanticLayerSource } from '../types.js';
|
import type { SemanticLayerSource } from '../types.js';
|
||||||
import { sourceDefinitionSchema } from './base-semantic-layer.tool.js';
|
import { sourceDefinitionSchema } from './base-semantic-layer.tool.js';
|
||||||
|
|
@ -28,7 +29,7 @@ function resolveDialect(warehouse: string | null): string | null {
|
||||||
if (!warehouse) {
|
if (!warehouse) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
return SemanticLayerService.mapDialect(warehouse);
|
return dialectForConnectionType(warehouse);
|
||||||
}
|
}
|
||||||
|
|
||||||
function wrapWithZeroRowQuery(sql: string, dialect: string): string {
|
function wrapWithZeroRowQuery(sql: string, dialect: string): string {
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,59 @@
|
||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
import { connectionTypeSchema } from '../../../src/context/connections/connection-type.js';
|
||||||
|
import {
|
||||||
|
dialectForConnectionType,
|
||||||
|
warehouseTargetDialect,
|
||||||
|
} from '../../../src/context/connections/connection-type-dialect.js';
|
||||||
|
|
||||||
|
describe('connection type dialect resolution', () => {
|
||||||
|
it('maps warehouse connection types to sqlglot dialects', () => {
|
||||||
|
const cases: Array<[string, string]> = [
|
||||||
|
['POSTGRESQL', 'postgres'],
|
||||||
|
['SQLITE', 'sqlite'],
|
||||||
|
['DUCKDB', 'duckdb'],
|
||||||
|
['SQLSERVER', 'tsql'],
|
||||||
|
['BIGQUERY', 'bigquery'],
|
||||||
|
['SNOWFLAKE', 'snowflake'],
|
||||||
|
['MYSQL', 'mysql'],
|
||||||
|
['CLICKHOUSE', 'clickhouse'],
|
||||||
|
['ATHENA', 'athena'],
|
||||||
|
];
|
||||||
|
|
||||||
|
for (const [connectionType, dialect] of cases) {
|
||||||
|
expect(dialectForConnectionType(connectionType)).toBe(dialect);
|
||||||
|
expect(warehouseTargetDialect(connectionType)).toBe(dialect);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it('normalizes case and preserves the semantic-layer postgres fallback for unknown inputs', () => {
|
||||||
|
expect(dialectForConnectionType('athena')).toBe('athena');
|
||||||
|
expect(dialectForConnectionType('postgresql')).toBe('postgres');
|
||||||
|
expect(dialectForConnectionType('not-a-real-connection-type')).toBe('postgres');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects non-SQL targets for BI table parsing', () => {
|
||||||
|
expect(warehouseTargetDialect('METABASE')).toBeNull();
|
||||||
|
expect(warehouseTargetDialect('LOOKER')).toBeNull();
|
||||||
|
expect(warehouseTargetDialect('NOTION')).toBeNull();
|
||||||
|
expect(warehouseTargetDialect('not-a-real-connection-type')).toBeNull();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('removes inherited non-ktx connection type values while keeping AWS Athena', () => {
|
||||||
|
expect(connectionTypeSchema.safeParse('ATHENA').success).toBe(true);
|
||||||
|
|
||||||
|
for (const removed of [
|
||||||
|
'CENTRALREACH',
|
||||||
|
'EPIC',
|
||||||
|
'CERNER',
|
||||||
|
'QUICKBOOKS',
|
||||||
|
'WORKDAY',
|
||||||
|
'REST',
|
||||||
|
'S3',
|
||||||
|
'SLACK',
|
||||||
|
'PLAIN',
|
||||||
|
'BETTERSTACK',
|
||||||
|
]) {
|
||||||
|
expect(connectionTypeSchema.safeParse(removed).success).toBe(false);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
@ -35,6 +35,21 @@ describe('localConnectionToWarehouseDescriptor', () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('maps Athena connections to canonical warehouse descriptors', () => {
|
||||||
|
expect(
|
||||||
|
localConnectionToWarehouseDescriptor('athena-warehouse', {
|
||||||
|
driver: 'athena',
|
||||||
|
region: 'us-east-1',
|
||||||
|
s3_staging_dir: 's3://my-bucket/athena-results/',
|
||||||
|
database: 'analytics',
|
||||||
|
}),
|
||||||
|
).toMatchObject({
|
||||||
|
id: 'athena-warehouse',
|
||||||
|
connection_type: 'ATHENA',
|
||||||
|
database: 'analytics',
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
it('returns null for non-warehouse adapters', () => {
|
it('returns null for non-warehouse adapters', () => {
|
||||||
expect(
|
expect(
|
||||||
localConnectionToWarehouseDescriptor('looker', {
|
localConnectionToWarehouseDescriptor('looker', {
|
||||||
|
|
@ -51,6 +66,7 @@ describe('local connection info helpers', () => {
|
||||||
expect(localConnectionTypeForConfig('warehouse', { driver: 'postgres' })).toBe('POSTGRESQL');
|
expect(localConnectionTypeForConfig('warehouse', { driver: 'postgres' })).toBe('POSTGRESQL');
|
||||||
expect(localConnectionTypeForConfig('bq', { driver: 'bigquery', project_id: 'acme' })).toBe('BIGQUERY');
|
expect(localConnectionTypeForConfig('bq', { driver: 'bigquery', project_id: 'acme' })).toBe('BIGQUERY');
|
||||||
expect(localConnectionTypeForConfig('snowflake', { driver: 'snowflake' })).toBe('SNOWFLAKE');
|
expect(localConnectionTypeForConfig('snowflake', { driver: 'snowflake' })).toBe('SNOWFLAKE');
|
||||||
|
expect(localConnectionTypeForConfig('athena-warehouse', { driver: 'athena' })).toBe('ATHENA');
|
||||||
});
|
});
|
||||||
|
|
||||||
it('keeps removed driver aliases as display-only labels', () => {
|
it('keeps removed driver aliases as display-only labels', () => {
|
||||||
|
|
|
||||||
|
|
@ -72,6 +72,8 @@ describe('looker dialect and target validation helpers', () => {
|
||||||
it('maps Looker dialect names to ktx connection types', () => {
|
it('maps Looker dialect names to ktx connection types', () => {
|
||||||
expect(lookerDialectToConnectionType('bigquery_standard_sql')).toBe('BIGQUERY');
|
expect(lookerDialectToConnectionType('bigquery_standard_sql')).toBe('BIGQUERY');
|
||||||
expect(lookerDialectToConnectionType('postgres')).toBe('POSTGRESQL');
|
expect(lookerDialectToConnectionType('postgres')).toBe('POSTGRESQL');
|
||||||
|
expect(lookerDialectToConnectionType('awsathena')).toBe('ATHENA');
|
||||||
|
expect(lookerDialectToConnectionType('AWSATHENA')).toBe('ATHENA');
|
||||||
expect(lookerDialectToConnectionType('mssql')).toBeNull();
|
expect(lookerDialectToConnectionType('mssql')).toBeNull();
|
||||||
expect(lookerDialectToConnectionType('tsql')).toBeNull();
|
expect(lookerDialectToConnectionType('tsql')).toBeNull();
|
||||||
expect(lookerDialectToConnectionType('unknown')).toBeNull();
|
expect(lookerDialectToConnectionType('unknown')).toBeNull();
|
||||||
|
|
@ -80,6 +82,7 @@ describe('looker dialect and target validation helpers', () => {
|
||||||
it('maps supported warehouse connection types to sqlglot dialects', () => {
|
it('maps supported warehouse connection types to sqlglot dialects', () => {
|
||||||
expect(sqlglotDialectForConnectionType('BIGQUERY')).toBe('bigquery');
|
expect(sqlglotDialectForConnectionType('BIGQUERY')).toBe('bigquery');
|
||||||
expect(sqlglotDialectForConnectionType('POSTGRESQL')).toBe('postgres');
|
expect(sqlglotDialectForConnectionType('POSTGRESQL')).toBe('postgres');
|
||||||
|
expect(sqlglotDialectForConnectionType('ATHENA')).toBe('athena');
|
||||||
expect(sqlglotDialectForConnectionType('LOOKER')).toBeNull();
|
expect(sqlglotDialectForConnectionType('LOOKER')).toBeNull();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
@ -259,6 +262,44 @@ describe('collectExploreParseItems and projectParsedIdentifier', () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('collects Athena parser inputs with the athena sqlglot dialect', () => {
|
||||||
|
expect(
|
||||||
|
collectExploreParseItems({
|
||||||
|
explore: {
|
||||||
|
...mappedExplore,
|
||||||
|
rawSqlTableName: 'analytics.orders',
|
||||||
|
joins: [
|
||||||
|
{
|
||||||
|
name: 'line_items',
|
||||||
|
type: 'left_outer',
|
||||||
|
relationship: 'one_to_many',
|
||||||
|
rawSqlTableName: 'analytics.line_items',
|
||||||
|
sqlOn: null,
|
||||||
|
from: null,
|
||||||
|
targetTable: null,
|
||||||
|
},
|
||||||
|
],
|
||||||
|
},
|
||||||
|
connectionMappings: { b2b_sandbox_bq: 'athena-warehouse' },
|
||||||
|
targetConnections: new Map([['athena-warehouse', { id: 'athena-warehouse', connection_type: 'ATHENA' }]]),
|
||||||
|
}),
|
||||||
|
).toEqual({
|
||||||
|
parsedTargetTables: {},
|
||||||
|
parseItems: [
|
||||||
|
{
|
||||||
|
key: 'b2b.sales_pipeline',
|
||||||
|
sql_table_name: 'analytics.orders',
|
||||||
|
dialect: 'athena',
|
||||||
|
},
|
||||||
|
{
|
||||||
|
key: 'b2b.sales_pipeline.line_items',
|
||||||
|
sql_table_name: 'analytics.line_items',
|
||||||
|
dialect: 'athena',
|
||||||
|
},
|
||||||
|
],
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
it('projects successful and failed parser rows into ktx parsed target tables', () => {
|
it('projects successful and failed parser rows into ktx parsed target tables', () => {
|
||||||
expect(
|
expect(
|
||||||
projectParsedIdentifier({
|
projectParsedIdentifier({
|
||||||
|
|
|
||||||
|
|
@ -121,6 +121,15 @@ describe('validateMappingPhysicalMatch', () => {
|
||||||
expect(reason).toContain('engine');
|
expect(reason).toContain('engine');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('accepts Athena mappings when the target connection type is ATHENA', () => {
|
||||||
|
expect(
|
||||||
|
validateMappingPhysicalMatch(
|
||||||
|
{ metabaseEngine: 'athena', metabaseDbName: 'analytics', metabaseHost: null },
|
||||||
|
{ connection_type: 'ATHENA', database: 'analytics' },
|
||||||
|
),
|
||||||
|
).toBeNull();
|
||||||
|
});
|
||||||
|
|
||||||
it('returns null when Postgres host and database both match after normalization', () => {
|
it('returns null when Postgres host and database both match after normalization', () => {
|
||||||
expect(
|
expect(
|
||||||
validateMappingPhysicalMatch(
|
validateMappingPhysicalMatch(
|
||||||
|
|
@ -159,6 +168,18 @@ describe('validateMappingPhysicalMatch', () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe('METABASE_ENGINE_TO_CONNECTION_TYPE', () => {
|
||||||
|
it('maps Metabase Athena databases to AWS Athena warehouse connections', () => {
|
||||||
|
expect(METABASE_ENGINE_TO_CONNECTION_TYPE.athena).toBe('ATHENA');
|
||||||
|
expect(
|
||||||
|
validateMappingPhysicalMatch(
|
||||||
|
{ metabaseEngine: 'athena', metabaseDbName: 'analytics', metabaseHost: null },
|
||||||
|
{ connection_type: 'POSTGRESQL', database: 'analytics' },
|
||||||
|
),
|
||||||
|
).toBe("Metabase database engine 'athena' does not match ktx connection type 'POSTGRESQL'");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe('computeMetabaseMappingPhysicalMismatches', () => {
|
describe('computeMetabaseMappingPhysicalMismatches', () => {
|
||||||
it('returns only mismatched physical mappings', () => {
|
it('returns only mismatched physical mappings', () => {
|
||||||
expect(
|
expect(
|
||||||
|
|
|
||||||
|
|
@ -960,6 +960,28 @@ describe('validateWithProposedSource', () => {
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('uses the Athena sqlglot dialect when validating Athena sources', async () => {
|
||||||
|
pythonPort.validateSources.mockResolvedValue({
|
||||||
|
data: { errors: [], warnings: [] },
|
||||||
|
});
|
||||||
|
service = new SemanticLayerService(configService as never, connectionCatalog('ATHENA'), pythonPort);
|
||||||
|
|
||||||
|
await service.validateWithProposedSource('conn-1', {
|
||||||
|
name: 'orders',
|
||||||
|
table: 'analytics.orders',
|
||||||
|
grain: ['id'],
|
||||||
|
columns: [{ name: 'id', type: 'number' }],
|
||||||
|
joins: [],
|
||||||
|
measures: [],
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(pythonPort.validateSources).toHaveBeenCalledWith(
|
||||||
|
expect.objectContaining({
|
||||||
|
dialect: 'athena',
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
it('composes a bare overlay with its manifest base before validating', async () => {
|
it('composes a bare overlay with its manifest base before validating', async () => {
|
||||||
const schemaPath = 'semantic-layer/conn-1/_schema/core.yaml';
|
const schemaPath = 'semantic-layer/conn-1/_schema/core.yaml';
|
||||||
const listFilesImpl = (dir: string): Promise<{ files: string[] }> => {
|
const listFilesImpl = (dir: string): Promise<{ files: string[] }> => {
|
||||||
|
|
@ -1326,6 +1348,44 @@ describe('validateWithProposedSource', () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe('executeQuery dialect resolution', () => {
|
||||||
|
it('passes the Athena sqlglot dialect to the Python query engine for Athena connections', async () => {
|
||||||
|
pythonPort.query.mockResolvedValue({ data: { sql: 'SELECT * FROM orders' } });
|
||||||
|
const catalog = connectionCatalog('ATHENA');
|
||||||
|
catalog.executeQuery.mockResolvedValue({ headers: ['id'], rows: [[1]], totalRows: 1 });
|
||||||
|
const configService = {
|
||||||
|
listFiles: vi.fn().mockResolvedValue({
|
||||||
|
files: ['semantic-layer/conn-1/orders.yaml'],
|
||||||
|
}),
|
||||||
|
readFile: vi.fn().mockResolvedValue({
|
||||||
|
content: [
|
||||||
|
'name: orders',
|
||||||
|
'table: analytics.orders',
|
||||||
|
'grain:',
|
||||||
|
' - id',
|
||||||
|
'columns:',
|
||||||
|
' - name: id',
|
||||||
|
' type: number',
|
||||||
|
'joins: []',
|
||||||
|
'measures:',
|
||||||
|
' - name: count',
|
||||||
|
' expr: count(*)',
|
||||||
|
].join('\n'),
|
||||||
|
}),
|
||||||
|
};
|
||||||
|
const service = new SemanticLayerService(configService as never, catalog, pythonPort);
|
||||||
|
|
||||||
|
await service.executeQuery('conn-1', { measures: ['count'] } as never);
|
||||||
|
|
||||||
|
expect(pythonPort.query).toHaveBeenCalledWith(
|
||||||
|
expect.objectContaining({
|
||||||
|
dialect: 'athena',
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
expect(catalog.executeQuery).toHaveBeenCalledWith('conn-1', 'SELECT * FROM orders');
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe('findDanglingSegmentRefs', () => {
|
describe('findDanglingSegmentRefs', () => {
|
||||||
it('returns empty when every measure segment resolves', () => {
|
it('returns empty when every measure segment resolves', () => {
|
||||||
const source = {
|
const source = {
|
||||||
|
|
|
||||||
|
|
@ -18,6 +18,8 @@ SUPPORTED_TABLE_IDENTIFIER_DIALECTS = {
|
||||||
"sqlite",
|
"sqlite",
|
||||||
"tsql",
|
"tsql",
|
||||||
"clickhouse",
|
"clickhouse",
|
||||||
|
"athena",
|
||||||
|
"duckdb",
|
||||||
}
|
}
|
||||||
|
|
||||||
ParseTableIdentifierReason = Literal[
|
ParseTableIdentifierReason = Literal[
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,16 @@ def test_parse_table_identifier_supported_dialects_and_aliases() -> None:
|
||||||
sql_table_name="RAW.PUBLIC.ORDERS",
|
sql_table_name="RAW.PUBLIC.ORDERS",
|
||||||
dialect="snowflake",
|
dialect="snowflake",
|
||||||
),
|
),
|
||||||
|
ParseTableIdentifierItem(
|
||||||
|
key="athena",
|
||||||
|
sql_table_name="analytics.orders",
|
||||||
|
dialect="athena",
|
||||||
|
),
|
||||||
|
ParseTableIdentifierItem(
|
||||||
|
key="duckdb",
|
||||||
|
sql_table_name="analytics.orders",
|
||||||
|
dialect="duckdb",
|
||||||
|
),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -37,6 +47,14 @@ def test_parse_table_identifier_supported_dialects_and_aliases() -> None:
|
||||||
assert response["sf"].catalog == "RAW"
|
assert response["sf"].catalog == "RAW"
|
||||||
assert response["sf"].schema_ == "PUBLIC"
|
assert response["sf"].schema_ == "PUBLIC"
|
||||||
assert response["sf"].name == "ORDERS"
|
assert response["sf"].name == "ORDERS"
|
||||||
|
assert response["athena"].ok is True
|
||||||
|
assert response["athena"].schema_ == "analytics"
|
||||||
|
assert response["athena"].name == "orders"
|
||||||
|
assert response["athena"].canonical_table == "analytics.orders"
|
||||||
|
assert response["duckdb"].ok is True
|
||||||
|
assert response["duckdb"].schema_ == "analytics"
|
||||||
|
assert response["duckdb"].name == "orders"
|
||||||
|
assert response["duckdb"].canonical_table == "analytics.orders"
|
||||||
|
|
||||||
|
|
||||||
def test_parse_table_identifier_rejects_non_physical_inputs() -> None:
|
def test_parse_table_identifier_rejects_non_physical_inputs() -> None:
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue