diff --git a/README.md b/README.md index 9b0d753..16d6e47 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,8 @@ cd packages/api && bun run scripts/service-matrix.ts | Analytics | MSK / Managed Streaming for Kafka | Yes (legacy page) | No | No | | Analytics | MSK Connect / Managed Streaming for Kafka Connect | Yes (legacy page) | No | No | | Analytics | Kinesis / Kinesis Data Streams | Yes (list, create, inspect, delete) | No | No | +| Analytics | Glue Data Catalog / Data Catalog | Yes (list, create, delete, inspect) | No | No | +| Analytics | Athena / Query | Yes (list, create, delete, inspect) | No | No | | Provisioning | CloudFormation / Infrastructure as Code | Yes (list, create, update, delete, inspect) | No | No | | Provisioning | AppConfig / Configuration | Yes (list, create, delete, inspect) | No | No | | Security | Identity | Yes (list, create, delete, inspect) | No | No | diff --git a/packages/api/package.json b/packages/api/package.json index d2ed3ab..231ba9a 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -14,6 +14,7 @@ "@aws-sdk/client-api-gateway": "^3.1076.0", "@aws-sdk/client-apigatewayv2": "^3.1076.0", "@aws-sdk/client-appconfig": "^3.1076.0", + "@aws-sdk/client-athena": "^3.1146.0", "@aws-sdk/client-auto-scaling": "^3.1146.0", "@aws-sdk/client-cloudformation": "^3.1111.0", "@aws-sdk/client-cloudfront": "^3.1146.0", @@ -32,6 +33,7 @@ "@aws-sdk/client-elastic-load-balancing-v2": "^3.1096.0", "@aws-sdk/client-elasticache": "^3.1146.0", "@aws-sdk/client-eventbridge": "^3.1076.0", + "@aws-sdk/client-glue": "^3.1146.0", "@aws-sdk/client-iam": "^3.1076.0", "@aws-sdk/client-kafka": "^3.1146.0", "@aws-sdk/client-kafkaconnect": "^3.1146.0", diff --git a/packages/api/src/adapter-aws/AwsDataCatalogAdapter.test.ts b/packages/api/src/adapter-aws/AwsDataCatalogAdapter.test.ts new file mode 100644 index 0000000..19cb11f --- /dev/null +++ b/packages/api/src/adapter-aws/AwsDataCatalogAdapter.test.ts @@ -0,0 +1,155 @@ +import {describe, expect, test} from 'bun:test' +import { + CreateDatabaseCommand, + CreateTableCommand, + DeleteDatabaseCommand, + DeleteTableCommand, + GetDatabaseCommand, + GetDatabasesCommand, + GetTablesCommand, + type GlueClient, +} from '@aws-sdk/client-glue' +import {AwsDataCatalogAdapter, describeTable} from './AwsDataCatalogAdapter' +import {NotFoundError, ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +const database = {Name: 'shop', Description: 'Orders and customers', LocationUri: 's3://data/shop/', CatalogId: '000000000000', CreateTime: new Date('2026-10-25T18:00:00Z')} +const ordersTable = { + Name: 'orders', + TableType: 'EXTERNAL_TABLE', + Parameters: {classification: 'csv', 'skip.header.line.count': '1'}, + StorageDescriptor: { + Columns: [{Name: 'id', Type: 'int'}, {Name: 'amount', Type: 'double'}], + Location: 's3://data/orders/', + InputFormat: 'org.apache.hadoop.mapred.TextInputFormat', + SerdeInfo: {SerializationLibrary: 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe', Parameters: {'field.delim': ';'}}, + }, + PartitionKeys: [{Name: 'day', Type: 'string'}], +} +const eventsTable = { + Name: 'events', + StorageDescriptor: {Columns: [{Name: 'kind', Type: 'string'}], Location: 's3://data/events/', SerdeInfo: {SerializationLibrary: 'org.openx.data.jsonserde.JsonSerDe'}}, +} + +function adapterWith(handler: (command: object) => SendResult | undefined = () => undefined) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + const result = handler(command) + if (result) return result + if (command instanceof GetDatabasesCommand) return {DatabaseList: [database]} + if (command instanceof GetDatabaseCommand) return {Database: database} + if (command instanceof GetTablesCommand) return {TableList: [ordersTable, eventsTable]} + return {} + }, + } + return {sent, adapter: new AwsDataCatalogAdapter(client as unknown as GlueClient)} +} + +describe('AwsDataCatalogAdapter', () => { + test('lists databases across pages with their table counts', async () => { + const {adapter} = adapterWith((command) => { + if (command instanceof GetDatabasesCommand) { + return command.input.NextToken ? {DatabaseList: [{Name: 'logs'}]} : {DatabaseList: [database], NextToken: 'n'} + } + if (command instanceof GetTablesCommand && command.input.DatabaseName === 'logs') return {TableList: []} + return undefined + }) + + const listed = await adapter.list() + expect(listed.map((db) => [db.name, db.metadata.tableCount, db.metadata.description])).toEqual([['shop', 2, 'Orders and customers'], ['logs', 0, null]]) + expect((await adapter.list({search: 'customers'})).map((db) => db.name)).toEqual(['shop']) + }) + + test('describes each table with its format, location and columns, sorted by name', async () => { + const {adapter} = adapterWith() + const detail = await adapter.get('shop') + const tables = detail?.metadata.tables as Array> + + expect(tables.map((table) => table.name)).toEqual(['events', 'orders']) + expect(tables[1]).toMatchObject({format: 'csv', delimiter: ';', headerRow: true, location: 's3://data/orders/', columns: [{name: 'id', type: 'int'}, {name: 'amount', type: 'double'}], partitionKeys: [{name: 'day', type: 'string'}]}) + expect(tables[0]).toMatchObject({format: 'json', delimiter: null, headerRow: false}) + }) + + test('reads the format the way the query engine does', () => { + expect(describeTable({Name: 'p', StorageDescriptor: {InputFormat: 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'}}).format).toBe('parquet') + expect(describeTable({Name: 'i', Parameters: {table_type: 'ICEBERG'}}).format).toBe('iceberg') + expect(describeTable({Name: 'c'}).format).toBe('csv') + }) + + test('returns null for a database that does not exist', async () => { + const missing = Object.assign(new Error('missing'), {name: 'EntityNotFoundException'}) + const {adapter} = adapterWith((command) => { if (command instanceof GetDatabaseCommand) throw missing; return undefined }) + expect(await adapter.get('nope')).toBeNull() + }) + + test('creates a database with an optional description and location', async () => { + const {adapter, sent} = adapterWith() + await adapter.create({values: {name: 'shop', description: 'Orders', locationUri: 's3://data/shop/'}}) + expect((sent.find((command) => command instanceof CreateDatabaseCommand) as CreateDatabaseCommand).input).toEqual({DatabaseInput: {Name: 'shop', Description: 'Orders', LocationUri: 's3://data/shop/'}}) + + await expect(adapter.create({values: {name: 'Shop'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {name: 'shop', locationUri: 'https://example.com'}})).rejects.toBeInstanceOf(ValidationError) + }) + + test('creates a CSV table Athena can read, with its delimiter and header row', async () => { + const {adapter, sent} = adapterWith() + await adapter.createCatalogTable('shop', {name: 'orders', location: 's3://data/orders', format: 'csv', delimiter: ';', headerRow: true, columns: [{name: 'id', type: 'int'}, {name: 'amount', type: 'DOUBLE'}]}) + + const create = sent.find((command) => command instanceof CreateTableCommand) as CreateTableCommand + expect(create.input).toEqual({ + DatabaseName: 'shop', + TableInput: { + Name: 'orders', + TableType: 'EXTERNAL_TABLE', + Parameters: {classification: 'csv', EXTERNAL: 'TRUE', 'skip.header.line.count': '1'}, + StorageDescriptor: { + Columns: [{Name: 'id', Type: 'int'}, {Name: 'amount', Type: 'double'}], + Location: 's3://data/orders/', + InputFormat: 'org.apache.hadoop.mapred.TextInputFormat', + OutputFormat: 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat', + SerdeInfo: {SerializationLibrary: 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe', Parameters: {'field.delim': ';'}}, + }, + }, + }) + }) + + test('creates JSON and Parquet tables with their own SerDe', async () => { + const {adapter, sent} = adapterWith() + await adapter.createCatalogTable('shop', {name: 'events', location: 's3://data/events/', format: 'json', columns: [{name: 'kind', type: 'string'}]}) + await adapter.createCatalogTable('shop', {name: 'facts', location: 's3://data/facts/', format: 'parquet', columns: [{name: 'total', type: 'decimal(12,2)'}]}) + + const [json, parquet] = sent.filter((command) => command instanceof CreateTableCommand) as CreateTableCommand[] + expect(json.input.TableInput?.StorageDescriptor?.SerdeInfo).toEqual({SerializationLibrary: 'org.openx.data.jsonserde.JsonSerDe'}) + expect(parquet.input.TableInput?.StorageDescriptor?.InputFormat).toBe('org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat') + expect(parquet.input.TableInput?.StorageDescriptor?.Columns).toEqual([{Name: 'total', Type: 'decimal(12,2)'}]) + }) + + test('refuses tables Athena could not query', async () => { + const {adapter} = adapterWith() + const table = {name: 'orders', location: 's3://data/orders/', format: 'csv' as const, columns: [{name: 'id', type: 'int'}]} + + await expect(adapter.createCatalogTable('shop', {...table, columns: []})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.createCatalogTable('shop', {...table, columns: [{name: 'id', type: 'int'}, {name: 'id', type: 'string'}]})).rejects.toThrow(/listed twice/) + await expect(adapter.createCatalogTable('shop', {...table, columns: [{name: 'id', type: 'uuid'}]})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.createCatalogTable('shop', {...table, location: '/tmp/orders'})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.createCatalogTable('shop', {...table, delimiter: '||'})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.createCatalogTable('shop', {...table, format: 'avro' as never})).rejects.toBeInstanceOf(ValidationError) + }) + + test('creating a table in a missing database is a not-found', async () => { + const missing = Object.assign(new Error('missing'), {name: 'EntityNotFoundException'}) + const {adapter} = adapterWith((command) => { if (command instanceof GetDatabaseCommand) throw missing; return undefined }) + await expect(adapter.createCatalogTable('nope', {name: 'orders', location: 's3://data/orders/', format: 'json', columns: [{name: 'id', type: 'int'}]})).rejects.toBeInstanceOf(NotFoundError) + }) + + test('deletes tables and databases', async () => { + const {adapter, sent} = adapterWith() + await adapter.deleteCatalogTable('shop', 'orders') + await adapter.delete('shop') + expect((sent.find((command) => command instanceof DeleteTableCommand) as DeleteTableCommand).input).toEqual({DatabaseName: 'shop', Name: 'orders'}) + expect((sent.find((command) => command instanceof DeleteDatabaseCommand) as DeleteDatabaseCommand).input).toEqual({Name: 'shop'}) + }) +}) diff --git a/packages/api/src/adapter-aws/AwsDataCatalogAdapter.ts b/packages/api/src/adapter-aws/AwsDataCatalogAdapter.ts new file mode 100644 index 0000000..730ef8c --- /dev/null +++ b/packages/api/src/adapter-aws/AwsDataCatalogAdapter.ts @@ -0,0 +1,257 @@ +import { + CreateDatabaseCommand, + CreateTableCommand, + type Database, + DeleteDatabaseCommand, + DeleteTableCommand, + GetDatabaseCommand, + GetDatabasesCommand, + GetTablesCommand, + type GlueClient, + type StorageDescriptor, + type Table, +} from '@aws-sdk/client-glue' +import {NotFoundError, RuntimeError, ValidationError} from '../cloud-spi/errors' +import {awsDataCatalogSchema, CATALOG_COLUMN_TYPES, CATALOG_NAME_PATTERN, CATALOG_TABLE_FORMATS} from '../cloud-spi/dataCatalogSchema' +import type {CatalogTableInput, CloudResource, CloudServiceAdapter, CreateResourceInput, ResourceQuery, ServiceSchema} from '../cloud-spi/types' + +const S3_URI = /^s3:\/\/[a-z0-9.-]{3,63}(\/.*)?$/ +const DECIMAL_TYPE = /^decimal\(\d{1,2},\d{1,2}\)$/ + +/** The Hive storage settings Athena expects for each format, as the Glue console writes them. */ +const STORAGE_FOR_FORMAT = { + csv: { + InputFormat: 'org.apache.hadoop.mapred.TextInputFormat', + OutputFormat: 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat', + SerializationLibrary: 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe', + }, + json: { + InputFormat: 'org.apache.hadoop.mapred.TextInputFormat', + OutputFormat: 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat', + SerializationLibrary: 'org.openx.data.jsonserde.JsonSerDe', + }, + parquet: { + InputFormat: 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat', + OutputFormat: 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat', + SerializationLibrary: 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe', + }, +} as const + +/** + * Talks to the AWS Glue Data Catalog, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - The catalog is what TheCloud's Athena reads: a table over CSV, JSON lines + * or Parquet files in S3 can be queried as soon as it exists. + * - Glue jobs and crawlers are stored but never run, so the console offers + * only the catalog: databases and the tables in them. + * - DeleteDatabase removes the database's tables too, as it does in AWS. + */ +export class AwsDataCatalogAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'datacatalog' as const + + constructor(private readonly glue: GlueClient) {} + + schema(): ServiceSchema { + return awsDataCatalogSchema() + } + + async list(query: ResourceQuery = {}): Promise { + const found: Database[] = [] + let next: string | undefined + do { + const res = await this.glue.send(new GetDatabasesCommand({MaxResults: 100, ...(next ? {NextToken: next} : {})})) + found.push(...(res.DatabaseList ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + const matching = found.filter((database) => !search + || (database.Name ?? '').toLowerCase().includes(search) + || (database.Description ?? '').toLowerCase().includes(search)) + return Promise.all(matching.map(async (database) => toResource(database, {tableCount: (await this.tables(database.Name ?? '')).length}))) + } + + async get(id: string): Promise { + const database = await this.database(id) + if (!database) return null + const tables = await this.tables(id) + return toResource(database, {tableCount: tables.length, tables}) + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = catalogName(values.name, 'name') + const description = text(values.description) + const locationUri = text(values.locationUri) + if (locationUri && !S3_URI.test(locationUri)) throw new ValidationError('locationUri must be an S3 URI such as s3://my-bucket/warehouse/') + + await this.glue.send(new CreateDatabaseCommand({ + DatabaseInput: { + Name: name, + ...(description ? {Description: description} : {}), + ...(locationUri ? {LocationUri: locationUri} : {}), + }, + })) + const created = await this.get(name) + if (!created) throw new RuntimeError(`Glue did not return the new database ${name}`) + return created + } + + async delete(id: string): Promise { + await this.glue.send(new DeleteDatabaseCommand({Name: id})) + } + + async createCatalogTable(databaseName: string, input: CatalogTableInput): Promise { + const name = catalogName(input.name, 'name') + const location = text(input.location) + if (!S3_URI.test(location)) throw new ValidationError('location must be an S3 URI such as s3://my-bucket/orders/') + const format = text(input.format) as CatalogTableInput['format'] + if (!(CATALOG_TABLE_FORMATS as readonly string[]).includes(format)) throw new ValidationError(`format must be one of ${CATALOG_TABLE_FORMATS.join(', ')}`) + const columns = tableColumns(input.columns) + const delimiter = typeof input.delimiter === 'string' && input.delimiter !== '' ? input.delimiter : ',' + if (format === 'csv' && [...delimiter].length !== 1) throw new ValidationError('delimiter must be a single character') + const description = text(input.description) + if (!await this.database(databaseName)) throw new NotFoundError(`Database ${databaseName} not found`) + + const storage = STORAGE_FOR_FORMAT[format] + const csv = format === 'csv' + await this.glue.send(new CreateTableCommand({ + DatabaseName: databaseName, + TableInput: { + Name: name, + TableType: 'EXTERNAL_TABLE', + ...(description ? {Description: description} : {}), + Parameters: { + classification: format, + EXTERNAL: 'TRUE', + ...(csv && input.headerRow ? {'skip.header.line.count': '1'} : {}), + }, + StorageDescriptor: { + Columns: columns.map((column) => ({Name: column.name, Type: column.type})), + // Athena reads every file under the prefix, so the location is a directory. + Location: location.endsWith('/') ? location : `${location}/`, + InputFormat: storage.InputFormat, + OutputFormat: storage.OutputFormat, + SerdeInfo: { + SerializationLibrary: storage.SerializationLibrary, + ...(csv ? {Parameters: {'field.delim': delimiter}} : {}), + }, + }, + }, + })) + return this.detail(databaseName) + } + + async deleteCatalogTable(databaseName: string, tableName: string): Promise { + await this.glue.send(new DeleteTableCommand({DatabaseName: databaseName, Name: tableName})) + return this.detail(databaseName) + } + + private async database(name: string): Promise { + try { + return (await this.glue.send(new GetDatabaseCommand({Name: name}))).Database ?? null + } catch (error) { + if (errorName(error) === 'EntityNotFoundException') return null + throw error + } + } + + private async detail(name: string): Promise { + const detail = await this.get(name) + if (!detail) throw new NotFoundError(`Database ${name} not found`) + return detail + } + + private async tables(databaseName: string): Promise { + const found: Table[] = [] + let next: string | undefined + do { + const res = await this.glue.send(new GetTablesCommand({DatabaseName: databaseName, MaxResults: 100, ...(next ? {NextToken: next} : {})})) + found.push(...(res.TableList ?? [])) + next = res.NextToken + } while (next) + return found.sort((a, b) => (a.Name ?? '').localeCompare(b.Name ?? '')) + } +} + +function toResource(database: Database, detail: {tableCount: number; tables?: Table[]}): CloudResource { + const name = database.Name ?? '' + return { + id: name, + name, + cloud: 'aws', + service: 'datacatalog', + type: 'glueDatabase', + region: null, + createdAt: database.CreateTime ? database.CreateTime.toISOString() : null, + metadata: { + description: database.Description || null, + locationUri: database.LocationUri || null, + catalogId: database.CatalogId ?? null, + tableCount: detail.tableCount, + ...(detail.tables ? {tables: detail.tables.map(describeTable)} : {}), + }, + } +} + +/** What the panel shows for a table: its files, how they are read, and its columns. */ +export function describeTable(table: Table) { + const storage: StorageDescriptor = table.StorageDescriptor ?? {} + const format = tableFormat(table) + return { + name: table.Name ?? '', + description: table.Description || null, + format, + location: storage.Location ?? null, + delimiter: format === 'csv' ? storage.SerdeInfo?.Parameters?.['field.delim'] ?? ',' : null, + headerRow: table.Parameters?.['skip.header.line.count'] === '1', + columns: (storage.Columns ?? []).map((column) => ({name: column.Name ?? '', type: column.Type ?? ''})), + partitionKeys: (table.PartitionKeys ?? []).map((column) => ({name: column.Name ?? '', type: column.Type ?? ''})), + tableType: table.TableType ?? null, + createdAt: table.CreateTime ? table.CreateTime.toISOString() : null, + updatedAt: table.UpdateTime ? table.UpdateTime.toISOString() : null, + } +} + +/** The same order the runtime's Athena engine checks a table's storage in. */ +function tableFormat(table: Table): string { + const classification = table.Parameters?.classification?.toLowerCase() + if (classification === 'iceberg' || table.Parameters?.table_type?.toLowerCase() === 'iceberg') return 'iceberg' + const storage = table.StorageDescriptor + const hints = `${storage?.InputFormat ?? ''} ${storage?.SerdeInfo?.SerializationLibrary ?? ''}`.toLowerCase() + if (hints.includes('parquet')) return 'parquet' + if (hints.includes('json')) return 'json' + return classification || 'csv' +} + +function tableColumns(value: unknown): Array<{name: string; type: string}> { + if (!Array.isArray(value) || value.length === 0) throw new ValidationError('columns must list at least one column') + const seen = new Set() + return value.map((column, index) => { + const entry = typeof column === 'object' && column !== null ? column as Record : {} + const name = catalogName(entry.name, `columns[${index}].name`) + if (seen.has(name)) throw new ValidationError(`column ${name} is listed twice`) + seen.add(name) + const type = text(entry.type).toLowerCase() + if (!(CATALOG_COLUMN_TYPES as readonly string[]).includes(type) && !DECIMAL_TYPE.test(type)) { + throw new ValidationError(`columns[${index}].type must be one of ${CATALOG_COLUMN_TYPES.join(', ')}`) + } + return {name, type} + }) +} + +function catalogName(value: unknown, field: string): string { + const name = text(value) + if (!new RegExp(CATALOG_NAME_PATTERN).test(name)) throw new ValidationError(`${field} must be lowercase letters, numbers and underscores, starting with a letter or underscore`) + return name +} + +function text(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} + +function errorName(error: unknown): string | undefined { + return typeof error === 'object' && error !== null ? (error as {name?: string}).name : undefined +} diff --git a/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts b/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts new file mode 100644 index 0000000..4ac7423 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts @@ -0,0 +1,166 @@ +import {describe, expect, test} from 'bun:test' +import { + type AthenaClient, + CreateWorkGroupCommand, + DeleteWorkGroupCommand, + GetQueryExecutionCommand, + GetQueryResultsCommand, + GetWorkGroupCommand, + ListDatabasesCommand, + ListQueryExecutionsCommand, + ListTableMetadataCommand, + ListWorkGroupsCommand, + StartQueryExecutionCommand, + StopQueryExecutionCommand, +} from '@aws-sdk/client-athena' +import {AwsQueryAdapter} from './AwsQueryAdapter' +import {NotFoundError, ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +function execution(id: string, overrides: Record = {}) { + return { + QueryExecutionId: id, + Query: 'SELECT name, amount FROM orders', + StatementType: 'DML', + WorkGroup: 'analysts', + QueryExecutionContext: {Database: 'shop', Catalog: 'AwsDataCatalog'}, + ResultConfiguration: {OutputLocation: `s3://results/${id}.csv`}, + Status: {State: 'SUCCEEDED', SubmissionDateTime: new Date('2026-10-25T21:00:00Z'), CompletionDateTime: new Date('2026-10-25T21:00:01Z')}, + Statistics: {TotalExecutionTimeInMillis: 42, DataScannedInBytes: 0}, + ...overrides, + } +} + +const header = {Data: [{VarCharValue: 'name'}, {VarCharValue: 'amount'}]} +const columnInfo = [{Name: 'name', Label: 'name', Type: 'varchar'}, {Name: 'amount', Label: 'amount', Type: 'double'}] +const row = (name: string, amount: string) => ({Data: [{VarCharValue: name}, {VarCharValue: amount}]}) + +function adapterWith(handler: (command: object) => SendResult | undefined = () => undefined) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + const result = handler(command) + if (result) return result + if (command instanceof GetQueryExecutionCommand) return {QueryExecution: execution(command.input.QueryExecutionId ?? '')} + if (command instanceof GetWorkGroupCommand) { + return {WorkGroup: {Name: command.input.WorkGroup, State: 'ENABLED', Description: 'Team', Configuration: {ResultConfiguration: {OutputLocation: 's3://results/'}, EnforceWorkGroupConfiguration: true, EngineVersion: {EffectiveEngineVersion: 'Athena engine version 3'}}}} + } + if (command instanceof StartQueryExecutionCommand) return {QueryExecutionId: 'q-new'} + return {} + }, + } + return {sent, adapter: new AwsQueryAdapter(client as unknown as AthenaClient)} +} + +describe('AwsQueryAdapter', () => { + test('lists workgroups across pages and describes one with its result location', async () => { + const {adapter} = adapterWith((command) => command instanceof ListWorkGroupsCommand + ? command.input.NextToken ? {WorkGroups: [{Name: 'analysts', State: 'ENABLED', Description: 'Team'}]} : {WorkGroups: [{Name: 'primary', State: 'ENABLED'}], NextToken: 'n'} + : undefined) + + expect((await adapter.list()).map((wg) => [wg.name, wg.status])).toEqual([['primary', 'ENABLED'], ['analysts', 'ENABLED']]) + expect((await adapter.list({search: 'team'})).map((wg) => wg.name)).toEqual(['analysts']) + expect((await adapter.get('analysts'))?.metadata).toEqual({description: 'Team', outputLocation: 's3://results/', enforceConfiguration: true, engineVersion: 'Athena engine version 3'}) + }) + + test('creates a workgroup with a result folder, and keeps the primary workgroup', async () => { + const {adapter, sent} = adapterWith() + await adapter.create({values: {name: 'analysts', description: 'Team', outputLocation: 's3://results/athena', enforce: 'true'}}) + expect((sent.find((command) => command instanceof CreateWorkGroupCommand) as CreateWorkGroupCommand).input).toEqual({ + Name: 'analysts', Description: 'Team', Configuration: {ResultConfiguration: {OutputLocation: 's3://results/athena/'}, EnforceWorkGroupConfiguration: true}, + }) + + await expect(adapter.create({values: {name: 'primary'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {name: 'bad name'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.delete('primary')).rejects.toBeInstanceOf(ValidationError) + await adapter.delete('analysts') + expect((sent.find((command) => command instanceof DeleteWorkGroupCommand) as DeleteWorkGroupCommand).input).toEqual({WorkGroup: 'analysts', RecursiveDeleteOption: true}) + }) + + test('starts a query in the workgroup against a database', async () => { + const {adapter, sent} = adapterWith() + const started = await adapter.startQuery('analysts', {sql: ' SELECT 1 ', database: 'shop'}) + + expect((sent.find((command) => command instanceof StartQueryExecutionCommand) as StartQueryExecutionCommand).input).toEqual({ + QueryString: 'SELECT 1', WorkGroup: 'analysts', QueryExecutionContext: {Catalog: 'AwsDataCatalog', Database: 'shop'}, + }) + expect(started).toMatchObject({id: 'q-new', state: 'SUCCEEDED', runtimeMs: 42, outputLocation: 's3://results/q-new.csv'}) + await expect(adapter.startQuery('analysts', {sql: ' '})).rejects.toBeInstanceOf(ValidationError) + }) + + test('returns a succeeded query with its columns and rows, without the header row', async () => { + const {adapter} = adapterWith((command) => command instanceof GetQueryResultsCommand + ? {ResultSet: {ResultSetMetadata: {ColumnInfo: columnInfo}, Rows: [header, row('alpha', '10'), row('beta', '25')]}} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.columns).toEqual([{name: 'name', type: 'varchar'}, {name: 'amount', type: 'double'}]) + expect(result.rows).toEqual([['alpha', '10'], ['beta', '25']]) + expect(result.truncated).toBe(false) + }) + + test('stops reading after the row limit and says the result is truncated', async () => { + const page = Array.from({length: 1000}, (_, i) => row(`n${i}`, String(i))) + const {adapter, sent} = adapterWith((command) => command instanceof GetQueryResultsCommand + ? command.input.NextToken ? {ResultSet: {Rows: page}, NextToken: 'more'} : {ResultSet: {ResultSetMetadata: {ColumnInfo: columnInfo}, Rows: [header, ...page.slice(0, 999)]}, NextToken: 'p2'} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.rows).toHaveLength(1000) + expect(result.truncated).toBe(true) + expect(sent.filter((command) => command instanceof GetQueryResultsCommand)).toHaveLength(2) + }) + + test('does not read results of a query that has not succeeded', async () => { + const {adapter, sent} = adapterWith((command) => command instanceof GetQueryExecutionCommand + ? {QueryExecution: execution('q1', {Status: {State: 'FAILED', StateChangeReason: 'Table orders not found'}})} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.execution).toMatchObject({state: 'FAILED', stateReason: 'Table orders not found'}) + expect(result.rows).toEqual([]) + expect(sent.some((command) => command instanceof GetQueryResultsCommand)).toBe(false) + }) + + test('refuses a query from another workgroup', async () => { + const {adapter, sent} = adapterWith() + await expect(adapter.getQueryResult('primary', 'q1')).rejects.toBeInstanceOf(NotFoundError) + await expect(adapter.stopQuery('primary', 'q1')).rejects.toBeInstanceOf(NotFoundError) + expect(sent.some((command) => command instanceof StopQueryExecutionCommand)).toBe(false) + + await adapter.stopQuery('analysts', 'q1') + expect(sent.some((command) => command instanceof StopQueryExecutionCommand)).toBe(true) + }) + + test('lists the workgroup queries newest first', async () => { + const {adapter} = adapterWith((command) => { + if (command instanceof ListQueryExecutionsCommand) return {QueryExecutionIds: ['old', 'new', 'other']} + if (command instanceof GetQueryExecutionCommand) { + const id = command.input.QueryExecutionId + return {QueryExecution: execution(id ?? '', { + WorkGroup: id === 'other' ? 'primary' : 'analysts', + Status: {State: 'SUCCEEDED', SubmissionDateTime: new Date(id === 'new' ? '2026-10-25T22:00:00Z' : '2026-10-25T21:00:00Z')}, + })} + } + return undefined + }) + + expect((await adapter.listQueries('analysts')).map((query) => query.id)).toEqual(['new', 'old']) + }) + + test('lists databases and the tables of the chosen one', async () => { + const {adapter, sent} = adapterWith((command) => { + if (command instanceof ListDatabasesCommand) return {DatabaseList: [{Name: 'shop'}, {Name: 'logs'}]} + if (command instanceof ListTableMetadataCommand) { + return {TableMetadataList: [{Name: 'orders', Columns: [{Name: 'id', Type: 'int'}], PartitionKeys: [{Name: 'day', Type: 'string'}]}]} + } + return undefined + }) + + expect(await adapter.describeQueryCatalog('shop')).toEqual({databases: ['logs', 'shop'], database: 'shop', tables: [{name: 'orders', columns: [{name: 'id', type: 'int'}, {name: 'day', type: 'string'}]}]}) + expect((await adapter.describeQueryCatalog('missing')).database).toBe('logs') + expect((sent.filter((command) => command instanceof ListTableMetadataCommand) as ListTableMetadataCommand[]).map((command) => command.input.DatabaseName)).toEqual(['shop', 'logs']) + }) +}) diff --git a/packages/api/src/adapter-aws/AwsQueryAdapter.ts b/packages/api/src/adapter-aws/AwsQueryAdapter.ts new file mode 100644 index 0000000..d288d47 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueryAdapter.ts @@ -0,0 +1,268 @@ +import { + type AthenaClient, + CreateWorkGroupCommand, + DeleteWorkGroupCommand, + GetQueryExecutionCommand, + GetQueryResultsCommand, + type GetQueryResultsCommandOutput, + GetWorkGroupCommand, + ListDatabasesCommand, + ListQueryExecutionsCommand, + ListTableMetadataCommand, + type ListTableMetadataCommandOutput, + ListWorkGroupsCommand, + type QueryExecution, + StartQueryExecutionCommand, + StopQueryExecutionCommand, + type TableMetadata, + type WorkGroup, + type WorkGroupSummary, +} from '@aws-sdk/client-athena' +import {NotFoundError, RuntimeError, ValidationError} from '../cloud-spi/errors' +import { + awsQuerySchema, + PRIMARY_WORKGROUP, + QUERY_HISTORY_LIMIT, + QUERY_RESULT_ROW_LIMIT, + WORKGROUP_NAME_PATTERN, +} from '../cloud-spi/querySchema' +import type { + CloudResource, + CloudServiceAdapter, + CreateResourceInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, + ResourceQuery, + ServiceSchema, +} from '../cloud-spi/types' + +/** Athena's own catalog name for the Glue Data Catalog. */ +const DATA_CATALOG = 'AwsDataCatalog' +/** Athena's cap on a query string. */ +const MAX_QUERY_LENGTH = 262_144 +const S3_URI = /^s3:\/\/[a-z0-9.-]{3,63}(\/.*)?$/ + +/** + * Talks to Amazon Athena, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - Queries really run: TheCloud executes the SQL with DuckDB over the files a + * Glue table points at, and writes the result to S3 as CSV, as Athena does. + * - Statements that return no rows, such as SHOW TABLES, fail in the runtime, + * so the editor lists databases and tables through the catalog APIs instead. + * - Data scanned is always reported as 0 bytes, so the editor shows only the + * run time. + */ +export class AwsQueryAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'query' as const + + constructor(private readonly athena: AthenaClient) {} + + schema(): ServiceSchema { + return awsQuerySchema() + } + + async list(query: ResourceQuery = {}): Promise { + const found: WorkGroupSummary[] = [] + let next: string | undefined + do { + const res = await this.athena.send(new ListWorkGroupsCommand({MaxResults: 50, ...(next ? {NextToken: next} : {})})) + found.push(...(res.WorkGroups ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + return found.map(toResource).filter((workGroup) => !search + || workGroup.name.toLowerCase().includes(search) + || String(workGroup.metadata.description ?? '').toLowerCase().includes(search)) + } + + async get(id: string): Promise { + try { + const workGroup = (await this.athena.send(new GetWorkGroupCommand({WorkGroup: id}))).WorkGroup + return workGroup ? toResource(workGroup) : null + } catch (error) { + if (errorName(error) === 'InvalidRequestException') return null + throw error + } + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = text(values.name) + if (!new RegExp(WORKGROUP_NAME_PATTERN).test(name)) throw new ValidationError('name must be up to 128 letters, numbers, dots, dashes or underscores') + if (name === PRIMARY_WORKGROUP) throw new ValidationError('Every account already has the primary workgroup') + const outputLocation = text(values.outputLocation) + if (outputLocation && !S3_URI.test(outputLocation)) throw new ValidationError('outputLocation must be an S3 URI such as s3://my-bucket/athena-results/') + const description = text(values.description) + + await this.athena.send(new CreateWorkGroupCommand({ + Name: name, + ...(description ? {Description: description} : {}), + Configuration: { + ...(outputLocation ? {ResultConfiguration: {OutputLocation: outputLocation.endsWith('/') ? outputLocation : `${outputLocation}/`}} : {}), + EnforceWorkGroupConfiguration: values.enforce === 'true', + }, + })) + const created = await this.get(name) + if (!created) throw new RuntimeError(`Athena did not return the new workgroup ${name}`) + return created + } + + async delete(id: string): Promise { + if (id === PRIMARY_WORKGROUP) throw new ValidationError('The primary workgroup cannot be deleted') + await this.athena.send(new DeleteWorkGroupCommand({WorkGroup: id, RecursiveDeleteOption: true})) + } + + async startQuery(workGroup: string, input: QueryInput): Promise { + const sql = typeof input.sql === 'string' ? input.sql.trim() : '' + if (!sql) throw new ValidationError('Enter a query to run') + if (sql.length > MAX_QUERY_LENGTH) throw new ValidationError(`A query can be at most ${MAX_QUERY_LENGTH} characters`) + const database = text(input.database) + + const started = await this.athena.send(new StartQueryExecutionCommand({ + QueryString: sql, + WorkGroup: workGroup, + ...(database ? {QueryExecutionContext: {Catalog: DATA_CATALOG, Database: database}} : {}), + })) + if (!started.QueryExecutionId) throw new RuntimeError('Athena did not return a query id') + return (await this.execution(workGroup, started.QueryExecutionId)) + } + + async getQueryResult(workGroup: string, queryId: string): Promise { + const execution = await this.execution(workGroup, queryId) + if (execution.state !== 'SUCCEEDED' || !execution.outputLocation) return {execution, columns: [], rows: [], truncated: false} + + const pages: GetQueryResultsCommandOutput[] = [] + let next: string | undefined + let rowCount = 0 + do { + const page = await this.athena.send(new GetQueryResultsCommand({QueryExecutionId: queryId, MaxResults: 1000, ...(next ? {NextToken: next} : {})})) + pages.push(page) + rowCount += page.ResultSet?.Rows?.length ?? 0 + next = page.NextToken + } while (next && rowCount <= QUERY_RESULT_ROW_LIMIT) + + const columnInfo = pages[0]?.ResultSet?.ResultSetMetadata?.ColumnInfo ?? [] + const raw = pages.flatMap((page) => page.ResultSet?.Rows ?? []).map((row) => (row.Data ?? []).map((cell) => cell.VarCharValue ?? null)) + const header = raw[0] ?? [] + const columns = columnInfo.length > 0 + ? columnInfo.map((column) => ({name: column.Label ?? column.Name ?? '', type: column.Type ?? null})) + : header.map((name) => ({name: name ?? '', type: null})) + // Athena repeats the column names as the first row of a SELECT's results. + const body = raw.length > 0 && header.length === columns.length && header.every((name, index) => name === columns[index].name) ? raw.slice(1) : raw + return {execution, columns, rows: body.slice(0, QUERY_RESULT_ROW_LIMIT), truncated: body.length > QUERY_RESULT_ROW_LIMIT || Boolean(next)} + } + + async stopQuery(workGroup: string, queryId: string): Promise { + await this.execution(workGroup, queryId) + await this.athena.send(new StopQueryExecutionCommand({QueryExecutionId: queryId})) + } + + async listQueries(workGroup: string): Promise { + const ids = (await this.athena.send(new ListQueryExecutionsCommand({WorkGroup: workGroup, MaxResults: QUERY_HISTORY_LIMIT}))).QueryExecutionIds ?? [] + const executions = await Promise.all(ids.slice(0, QUERY_HISTORY_LIMIT).map(async (id) => { + const res = await this.athena.send(new GetQueryExecutionCommand({QueryExecutionId: id})) + return res.QueryExecution ? toExecution(res.QueryExecution) : null + })) + return executions + .filter((execution): execution is QueryExecutionSummary => execution !== null && (execution.workGroup ?? PRIMARY_WORKGROUP) === workGroup) + .sort((a, b) => (b.submittedAt ?? '').localeCompare(a.submittedAt ?? '')) + } + + async describeQueryCatalog(database?: string): Promise { + const databases: string[] = [] + let next: string | undefined + do { + const res = await this.athena.send(new ListDatabasesCommand({CatalogName: DATA_CATALOG, MaxResults: 50, ...(next ? {NextToken: next} : {})})) + databases.push(...(res.DatabaseList ?? []).map((entry) => entry.Name ?? '').filter(Boolean)) + next = res.NextToken + } while (next) + databases.sort() + + const chosen = database && databases.includes(database) ? database : databases[0] ?? null + if (!chosen) return {databases, database: null, tables: []} + + const tables: TableMetadata[] = [] + let tableToken: string | undefined + do { + const res: ListTableMetadataCommandOutput = await this.athena.send(new ListTableMetadataCommand({CatalogName: DATA_CATALOG, DatabaseName: chosen, MaxResults: 50, ...(tableToken ? {NextToken: tableToken} : {})})) + tables.push(...(res.TableMetadataList ?? [])) + tableToken = res.NextToken + } while (tableToken) + return { + databases, + database: chosen, + tables: tables.map((table) => ({ + name: table.Name ?? '', + columns: [...(table.Columns ?? []), ...(table.PartitionKeys ?? [])].map((column) => ({name: column.Name ?? '', type: column.Type ?? ''})), + })).sort((a, b) => a.name.localeCompare(b.name)), + } + } + + /** A query, refused unless it ran in this workgroup, so one workgroup's page never shows another's. */ + private async execution(workGroup: string, queryId: string): Promise { + let execution: QueryExecution | undefined + try { + execution = (await this.athena.send(new GetQueryExecutionCommand({QueryExecutionId: queryId}))).QueryExecution + } catch (error) { + if (errorName(error) === 'InvalidRequestException') throw new NotFoundError(`Query ${queryId} not found`) + throw error + } + if (!execution || (execution.WorkGroup ?? PRIMARY_WORKGROUP) !== workGroup) throw new NotFoundError(`Query ${queryId} is not in the ${workGroup} workgroup`) + return toExecution(execution) + } +} + +function toResource(workGroup: WorkGroup | WorkGroupSummary): CloudResource { + const name = workGroup.Name ?? '' + const detail = 'Configuration' in workGroup ? workGroup : undefined + const configuration = detail?.Configuration + return { + id: name, + name, + cloud: 'aws', + service: 'query', + type: 'athenaWorkGroup', + region: null, + createdAt: workGroup.CreationTime ? workGroup.CreationTime.toISOString() : null, + status: workGroup.State ?? null, + metadata: { + description: workGroup.Description || null, + ...(detail ? { + outputLocation: configuration?.ResultConfiguration?.OutputLocation ?? null, + enforceConfiguration: configuration?.EnforceWorkGroupConfiguration === true, + engineVersion: configuration?.EngineVersion?.EffectiveEngineVersion ?? configuration?.EngineVersion?.SelectedEngineVersion ?? null, + } : {}), + }, + } +} + +function toExecution(execution: QueryExecution): QueryExecutionSummary { + const status = execution.Status + return { + id: execution.QueryExecutionId ?? '', + sql: execution.Query ?? '', + database: execution.QueryExecutionContext?.Database ?? null, + workGroup: execution.WorkGroup ?? null, + state: status?.State ?? 'QUEUED', + stateReason: status?.StateChangeReason ?? status?.AthenaError?.ErrorMessage ?? null, + statementType: execution.StatementType ?? null, + submittedAt: status?.SubmissionDateTime ? status.SubmissionDateTime.toISOString() : null, + completedAt: status?.CompletionDateTime ? status.CompletionDateTime.toISOString() : null, + runtimeMs: execution.Statistics?.TotalExecutionTimeInMillis ?? execution.Statistics?.EngineExecutionTimeInMillis ?? null, + dataScannedBytes: execution.Statistics?.DataScannedInBytes ?? null, + outputLocation: execution.ResultConfiguration?.OutputLocation ?? null, + } +} + +function text(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} + +function errorName(error: unknown): string | undefined { + return typeof error === 'object' && error !== null ? (error as {name?: string}).name : undefined +} diff --git a/packages/api/src/aws.ts b/packages/api/src/aws.ts index 5ca46c1..6e9ba28 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -38,6 +38,8 @@ import { AutoScalingClient } from "@aws-sdk/client-auto-scaling"; import { ACMClient } from "@aws-sdk/client-acm"; import { PipesClient } from "@aws-sdk/client-pipes"; import { EFSClient } from "@aws-sdk/client-efs"; +import { GlueClient } from "@aws-sdk/client-glue"; +import { AthenaClient } from "@aws-sdk/client-athena"; const endpoint = process.env.FLOCI_ENDPOINT; const region = process.env.AWS_REGION || "us-east-1"; @@ -107,6 +109,8 @@ export type AwsClients = { acm: ACMClient; pipes: PipesClient; efs: EFSClient; + glue: GlueClient; + athena: AthenaClient; }; export type AwsClientName = keyof AwsClients; @@ -160,6 +164,8 @@ function buildClients(accountId: string): AwsClients { acm: new ACMClient(base), pipes: new PipesClient(base), efs: new EFSClient(base), + glue: new GlueClient(base), + athena: new AthenaClient(base), }; } diff --git a/packages/api/src/cloud-spi/dataCatalogSchema.ts b/packages/api/src/cloud-spi/dataCatalogSchema.ts new file mode 100644 index 0000000..2103efe --- /dev/null +++ b/packages/api/src/cloud-spi/dataCatalogSchema.ts @@ -0,0 +1,68 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** + * Glue's rule for database and table names. Glue lowercases them, and Athena + * needs them to start with a letter or underscore to query them unquoted. + */ +export const CATALOG_NAME_PATTERN = '^[a-z_][a-z0-9_]{0,254}$' +/** The data formats TheCloud's Athena engine reads from a table's S3 location. */ +export const CATALOG_TABLE_FORMATS = ['csv', 'json', 'parquet'] as const +/** Hive column types Athena understands and the console offers. */ +export const CATALOG_COLUMN_TYPES = ['string', 'int', 'bigint', 'double', 'float', 'boolean', 'date', 'timestamp', 'decimal(10,2)'] as const + +const databaseColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'tables', label: 'Tables', path: 'metadata.tableCount', emptyText: '—'}, + {name: 'description', label: 'Description', path: 'metadata.description', emptyText: '—'}, + {name: 'location', label: 'Location', path: 'metadata.locationUri', emptyText: '—'}, + {name: 'createdAt', label: 'Created', format: 'datetime'}, +] + +const databaseFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +export function awsDataCatalogSchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'datacatalog', + displayName: 'Glue Data Catalog databases', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + description: 'Lowercase letters, numbers and underscores, starting with a letter or underscore, so Athena can query it.', + validation: {pattern: CATALOG_NAME_PATTERN, message: 'Use lowercase letters, numbers and underscores, starting with a letter or underscore'}, + }, + {name: 'description', label: 'Description', type: 'text', required: false, validation: {maxLength: 2048}}, + { + name: 'locationUri', + label: 'Location', + type: 'text', + required: false, + span: true, + description: 'Optional S3 location for the database, such as s3://my-bucket/warehouse/.', + validation: {pattern: '^s3://[a-z0-9.-]{3,63}(/.*)?$', message: 'Use an S3 URI such as s3://my-bucket/warehouse/'}, + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List databases', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create database', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete database', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect database', enabled: true, status: 'available', runtimeRequired: true}, + ], + dataCatalogActions: [ + {name: 'createTable', label: 'Create table', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'deleteTable', label: 'Delete table', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: databaseFilters, + columns: databaseColumns, + } +} + +export function dataCatalogSchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsDataCatalogSchema() : null +} diff --git a/packages/api/src/cloud-spi/querySchema.ts b/packages/api/src/cloud-spi/querySchema.ts new file mode 100644 index 0000000..f9071d0 --- /dev/null +++ b/packages/api/src/cloud-spi/querySchema.ts @@ -0,0 +1,80 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** CreateWorkGroup's name rule. */ +export const WORKGROUP_NAME_PATTERN = '^[a-zA-Z0-9._-]{1,128}$' +/** The workgroup every account has, which Athena does not let you delete. */ +export const PRIMARY_WORKGROUP = 'primary' +/** Rows the editor shows for one query; the full result stays in the CSV file in S3. */ +export const QUERY_RESULT_ROW_LIMIT = 1000 +/** Recent queries the history shows. */ +export const QUERY_HISTORY_LIMIT = 25 + +const workGroupColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'status', label: 'State', format: 'badge'}, + {name: 'description', label: 'Description', path: 'metadata.description', emptyText: '—'}, + {name: 'createdAt', label: 'Created', format: 'datetime', emptyText: '—'}, +] + +const workGroupFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +export function awsQuerySchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'query', + displayName: 'Athena workgroups', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + validation: {pattern: WORKGROUP_NAME_PATTERN, message: 'Use up to 128 letters, numbers, dots, dashes and underscores'}, + }, + {name: 'description', label: 'Description', type: 'text', required: false, validation: {maxLength: 1024}}, + { + name: 'outputLocation', + label: 'Query result location', + type: 'text', + required: false, + span: true, + description: 'Where each query writes its results as a CSV file, such as s3://my-bucket/athena-results/. Leave empty to use TheCloud\'s default results bucket.', + validation: {pattern: '^s3://[a-z0-9.-]{3,63}(/.*)?$', message: 'Use an S3 URI such as s3://my-bucket/athena-results/'}, + }, + { + name: 'enforce', + label: 'Override client-side settings', + type: 'select', + required: false, + defaultValue: 'false', + description: 'When on, queries always use this workgroup\'s result location, even if a client asks for another.', + options: [ + {label: 'Off', value: 'false'}, + {label: 'On', value: 'true'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List workgroups', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create workgroup', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete workgroup', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect workgroup', enabled: true, status: 'available', runtimeRequired: true}, + ], + queryActions: [ + {name: 'run', label: 'Run query', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'results', label: 'Get query results', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'stop', label: 'Cancel query', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'history', label: 'Recent queries', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'catalog', label: 'Browse databases and tables', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: workGroupFilters, + columns: workGroupColumns, + } +} + +export function querySchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsQuerySchema() : null +} diff --git a/packages/api/src/cloud-spi/serviceCatalog.ts b/packages/api/src/cloud-spi/serviceCatalog.ts index 616759d..1e0eeda 100644 --- a/packages/api/src/cloud-spi/serviceCatalog.ts +++ b/packages/api/src/cloud-spi/serviceCatalog.ts @@ -222,6 +222,20 @@ export const SERVICE_CATALOG = { group: 'Analytics', order: 18, }, + datacatalog: { + displayName: 'Data Catalog', + displayNameByCloud: {aws: 'Glue Data Catalog'}, + iconKey: 'datacatalog', + group: 'Analytics', + order: 20, + }, + query: { + displayName: 'Query', + displayNameByCloud: {aws: 'Athena'}, + iconKey: 'query', + group: 'Analytics', + order: 21, + }, identity: {displayName: 'Identity', iconKey: 'iam', group: 'Security', order: 5}, apigateway: {displayName: 'API Gateway', iconKey: 'apigateway', group: 'Integration', order: 10}, email: { diff --git a/packages/api/src/cloud-spi/types.ts b/packages/api/src/cloud-spi/types.ts index a4e2de0..d8188c7 100644 --- a/packages/api/src/cloud-spi/types.ts +++ b/packages/api/src/cloud-spi/types.ts @@ -188,6 +188,8 @@ export type UserPoolActionName = export type DistributionActionName = 'setEnabled' | 'createInvalidation' export type AutoScalingActionName = 'setCapacity' | 'terminateInstance' export type CertificateActionName = 'download' | 'export' | 'revoke' +export type QueryActionName = 'run' | 'results' | 'stop' | 'history' | 'catalog' +export type DataCatalogActionName = 'createTable' | 'deleteTable' export type FileSystemActionName = 'createMountTarget' | 'deleteMountTarget' | 'createAccessPoint' | 'deleteAccessPoint' | 'attachFunction' | 'detachFunction' export type MonitoringActionName = 'listMetrics' | 'getMetricData' | 'alarmHistory' | 'setAlarmState' | 'setAlarmActions' export type HttpApiActionName = @@ -251,6 +253,8 @@ export interface ServiceSchema { autoScalingActions?: CapabilitySchema[] certificateActions?: CapabilitySchema[] fileSystemActions?: CapabilitySchema[] + dataCatalogActions?: CapabilitySchema[] + queryActions?: CapabilitySchema[] monitoringActions?: CapabilitySchema[] /** API Gateway v2 (HTTP and WebSocket APIs): what the routes, stages and security panel can do. */ httpApiActions?: CapabilitySchema[] @@ -558,6 +562,61 @@ export interface AccessPointInput { permissions: string } +export interface QueryInput { + sql: string + /** The Glue database unqualified table names resolve in. */ + database?: string +} + +export interface QueryExecutionSummary { + id: string + sql: string + database: string | null + workGroup: string | null + /** QUEUED, RUNNING, SUCCEEDED, FAILED or CANCELLED. */ + state: string + stateReason: string | null + statementType: string | null + submittedAt: string | null + completedAt: string | null + runtimeMs: number | null + dataScannedBytes: number | null + /** The S3 object holding the full result as CSV. */ + outputLocation: string | null +} + +export interface QueryResult { + execution: QueryExecutionSummary + columns: Array<{name: string; type: string | null}> + rows: Array> + /** More rows exist than the console returns; the full result is in the output file. */ + truncated: boolean +} + +export interface QueryCatalog { + databases: string[] + database: string | null + tables: Array<{name: string; columns: Array<{name: string; type: string}>}> +} + +export interface CatalogColumnInput { + name: string + type: string +} + +export interface CatalogTableInput { + name: string + description?: string + /** The S3 prefix holding the table's files, such as s3://bucket/orders/. */ + location: string + format: 'csv' | 'json' | 'parquet' + /** CSV only: the field separator, a comma unless given. */ + delimiter?: string + /** CSV only: whether the first line of each file is a header to skip. */ + headerRow?: boolean + columns: CatalogColumnInput[] +} + export interface FileSystemAttachmentInput { accessPointId: string functionName: string @@ -1235,6 +1294,15 @@ export interface CloudServiceAdapter { /** With `replace`, the group launches a new instance in its place; otherwise desired capacity drops by one. */ terminateAutoScalingInstance?(groupName: string, instanceId: string, replace: boolean): Promise getCertificatePem?(certificateId: string): Promise + startQuery?(workGroup: string, input: QueryInput): Promise + /** The query's state, and its first rows once it has succeeded. */ + getQueryResult?(workGroup: string, queryId: string): Promise + stopQuery?(workGroup: string, queryId: string): Promise + listQueries?(workGroup: string): Promise + /** The databases a query can use and, for the one named, its tables. */ + describeQueryCatalog?(database?: string): Promise + createCatalogTable?(databaseName: string, input: CatalogTableInput): Promise + deleteCatalogTable?(databaseName: string, tableName: string): Promise createMountTarget?(fileSystemId: string, subnetId: string): Promise deleteMountTarget?(fileSystemId: string, mountTargetId: string): Promise createAccessPoint?(fileSystemId: string, input: AccessPointInput): Promise diff --git a/packages/api/src/cloudProxy.test.ts b/packages/api/src/cloudProxy.test.ts index 58a4b6d..9331415 100644 --- a/packages/api/src/cloudProxy.test.ts +++ b/packages/api/src/cloudProxy.test.ts @@ -2,7 +2,7 @@ import {describe, expect, test} from 'bun:test' import {createCloudAdapterRegistry} from './cloudProxy' import {checkChildCapabilities} from './cloud-spi/childCapabilities' import {isServiceType} from './cloud-spi/serviceCatalog' -import type {AutoScalingActionName, CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DistributionActionName, FileSystemActionName, HttpApiActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' +import type {AutoScalingActionName, CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DataCatalogActionName, DistributionActionName, FileSystemActionName, HttpApiActionName, QueryActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' /** * Guards the schema-to-implementation contract for every registered adapter. @@ -81,6 +81,19 @@ const METHOD_FOR_CERTIFICATE_ACTION: Record = { + run: 'startQuery', + results: 'getQueryResult', + stop: 'stopQuery', + history: 'listQueries', + catalog: 'describeQueryCatalog', +} + +const METHOD_FOR_DATA_CATALOG_ACTION: Record = { + createTable: 'createCatalogTable', + deleteTable: 'deleteCatalogTable', +} + const METHOD_FOR_FILE_SYSTEM_ACTION: Record = { createMountTarget: 'createMountTarget', deleteMountTarget: 'deleteMountTarget', @@ -226,6 +239,24 @@ describe('registered adapters honour their schema', () => { ).toBe('function') } + const queryClaims = (adapter.schema().capabilities?.queryActions ?? []).filter((capability) => capability.status === 'available') + for (const capability of queryClaims) { + const method = METHOD_FOR_QUERY_ACTION[capability.name] + expect( + typeof adapter[method], + `${label} advertises ${capability.name} as available but has no ${String(method)}()`, + ).toBe('function') + } + + const dataCatalogClaims = (adapter.schema().capabilities?.dataCatalogActions ?? []).filter((capability) => capability.status === 'available') + for (const capability of dataCatalogClaims) { + const method = METHOD_FOR_DATA_CATALOG_ACTION[capability.name] + expect( + typeof adapter[method], + `${label} advertises ${capability.name} as available but has no ${String(method)}()`, + ).toBe('function') + } + const fileSystemClaims = (adapter.schema().capabilities?.fileSystemActions ?? []).filter((capability) => capability.status === 'available') for (const capability of fileSystemClaims) { const method = METHOD_FOR_FILE_SYSTEM_ACTION[capability.name] @@ -259,6 +290,8 @@ describe('registered adapters honour their schema', () => { ...(adapter.schema().capabilities?.distributionActions ?? []), ...(adapter.schema().capabilities?.autoScalingActions ?? []), ...(adapter.schema().capabilities?.fileSystemActions ?? []), + ...(adapter.schema().capabilities?.dataCatalogActions ?? []), + ...(adapter.schema().capabilities?.queryActions ?? []), ...(adapter.schema().capabilities?.certificateActions ?? []), ...(adapter.schema().capabilities?.monitoringActions ?? []), ...(adapter.schema().capabilities?.httpApiActions ?? []), diff --git a/packages/api/src/cloudProxy.ts b/packages/api/src/cloudProxy.ts index 925c261..dd5b44d 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -54,6 +54,8 @@ import {AwsAutoScalingAdapter} from './adapter-aws/AwsAutoScalingAdapter' import {AwsCertificatesAdapter} from './adapter-aws/AwsCertificatesAdapter' import {AwsPipesAdapter} from './adapter-aws/AwsPipesAdapter' import {AwsFileSystemsAdapter} from './adapter-aws/AwsFileSystemsAdapter' +import {AwsDataCatalogAdapter} from './adapter-aws/AwsDataCatalogAdapter' +import {AwsQueryAdapter} from './adapter-aws/AwsQueryAdapter' import {AwsMonitoringAdapter} from './adapter-aws/AwsMonitoringAdapter' import {awsClientsForAccount, resolveAccountId} from './aws' import {createEc2Service} from './services/ec2' @@ -111,6 +113,8 @@ export function createCloudAdapterRegistry(accountId?: string | null): CloudAdap new AwsCertificatesAdapter(clients.acm, resolveAccountId(accountId)), new AwsPipesAdapter(clients.pipes), new AwsFileSystemsAdapter(clients.efs, clients.lambda, clients.ec2), + new AwsDataCatalogAdapter(clients.glue), + new AwsQueryAdapter(clients.athena), new AwsMonitoringAdapter(clients.cloudwatch), new AzureStorageAdapter(), new AzureServiceBusAdapter(), diff --git a/packages/api/src/routes/clouds.test.ts b/packages/api/src/routes/clouds.test.ts index 546b58a..919ce3d 100644 --- a/packages/api/src/routes/clouds.test.ts +++ b/packages/api/src/routes/clouds.test.ts @@ -20,6 +20,8 @@ import {awsUserPoolsSchema} from '../cloud-spi/userPoolsSchema' import {awsCdnSchema} from '../cloud-spi/cdnSchema' import {awsAutoScalingSchema} from '../cloud-spi/autoScalingSchema' import {awsFileSystemsSchema} from '../cloud-spi/fileSystemsSchema' +import {awsDataCatalogSchema} from '../cloud-spi/dataCatalogSchema' +import {awsQuerySchema} from '../cloud-spi/querySchema' import {awsCertificatesSchema} from '../cloud-spi/certificatesSchema' import {awsMonitoringSchema} from '../cloud-spi/monitoringSchema' import {awsAppConfigSchema} from '../cloud-spi/appConfigSchema' @@ -876,6 +878,45 @@ describe('cloud schema routes', () => { expect(calls).toEqual([['capacity', 'web-asg', {desiredCapacity: 3}], ['terminate', 'web-asg', 'i-abc', true]]) }) + test('runs, reads, lists and cancels queries through the Athena adapter', async () => { + const calls: unknown[] = [] + const execution = {id: 'q1', sql: 'SELECT 1', database: 'shop', workGroup: 'analysts', state: 'SUCCEEDED', stateReason: null, statementType: 'DML', submittedAt: null, completedAt: null, runtimeMs: 5, dataScannedBytes: 0, outputLocation: 's3://r/q1.csv'} + const app = appWithRoutes([mockAdapter('aws', { + service: 'query', + schema: awsQuerySchema, + startQuery: async (id, input) => { calls.push(['start', id, input.sql, input.database]); return execution }, + getQueryResult: async (id, queryId) => { calls.push(['result', id, queryId]); return {execution, columns: [{name: '_col0', type: 'integer'}], rows: [['1']], truncated: false} }, + stopQuery: async (id, queryId) => { calls.push(['stop', id, queryId]) }, + listQueries: async (id) => { calls.push(['list', id]); return [execution] }, + describeQueryCatalog: async (database) => { calls.push(['catalog', database]); return {databases: ['shop'], database: 'shop', tables: []} }, + })]) + const base = '/api/clouds/aws/services/query/resources/analysts/queries' + + expect((await app.request(base, {method: 'POST', headers: {'content-type': 'application/json'}, body: JSON.stringify({sql: 'SELECT 1', database: 'shop'})})).status).toBe(201) + expect((await (await app.request(`${base}/q1`)).json()).rows).toEqual([['1']]) + expect((await app.request(base)).status).toBe(200) + expect((await app.request(`${base}/q1/stop`, {method: 'POST'})).status).toBe(200) + expect((await (await app.request('/api/clouds/aws/services/query/catalog?database=shop')).json()).databases).toEqual(['shop']) + expect(calls).toEqual([['start', 'analysts', 'SELECT 1', 'shop'], ['result', 'analysts', 'q1'], ['list', 'analysts'], ['stop', 'analysts', 'q1'], ['catalog', 'shop']]) + }) + + test('creates and deletes catalog tables through the Glue adapter', async () => { + const calls: unknown[] = [] + const database = {id: 'shop', name: 'shop', cloud: 'aws' as const, service: 'datacatalog' as const, type: 'glueDatabase', region: null, createdAt: null, metadata: {}} + const app = appWithRoutes([mockAdapter('aws', { + service: 'datacatalog', + schema: awsDataCatalogSchema, + createCatalogTable: async (id, input) => { calls.push(['create', id, input.name, input.format]); return database }, + deleteCatalogTable: async (id, tableName) => { calls.push(['delete', id, tableName]); return database }, + })]) + const base = '/api/clouds/aws/services/datacatalog/resources/shop/tables' + + const created = await app.request(base, {method: 'POST', headers: {'content-type': 'application/json'}, body: JSON.stringify({name: 'orders', location: 's3://data/orders/', format: 'csv', columns: [{name: 'id', type: 'int'}]})}) + expect(created.status).toBe(201) + expect((await app.request(`${base}/orders`, {method: 'DELETE'})).status).toBe(200) + expect(calls).toEqual([['create', 'shop', 'orders', 'csv'], ['delete', 'shop', 'orders']]) + }) + test('manages mount targets, access points and function mounts through the EFS adapter', async () => { const calls: unknown[] = [] const fileSystem = {id: 'fs-1', name: 'shared', cloud: 'aws' as const, service: 'filesystems' as const, type: 'fileSystem', region: null, createdAt: null, metadata: {}} diff --git a/packages/api/src/routes/clouds.ts b/packages/api/src/routes/clouds.ts index f879e5d..ecd484b 100644 --- a/packages/api/src/routes/clouds.ts +++ b/packages/api/src/routes/clouds.ts @@ -4,6 +4,8 @@ import type { AlarmStateValue, AutoScalingCapacityInput, AccessPointInput, + CatalogTableInput, + QueryInput, FileSystemAttachmentInput, CertificateRevocationReason, CloudProvider, @@ -556,6 +558,58 @@ export function createCloudRoutes(injectedService?: CloudProxyService) { }) }) + app.get('/:cloud/services/query/catalog', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).describeQueryCatalog(cloud, c.req.query('database') || undefined))) + }) + + app.get('/:cloud/services/query/resources/:id/queries', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).listQueries(cloud, c.req.param('id')))) + }) + + app.post('/:cloud/services/query/resources/:id/queries', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).startQuery(cloud, c.req.param('id'), await c.req.json()), 201)) + }) + + app.get('/:cloud/services/query/resources/:id/queries/:queryId', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).getQueryResult(cloud, c.req.param('id'), c.req.param('queryId')))) + }) + + app.post('/:cloud/services/query/resources/:id/queries/:queryId/stop', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => { + await svc(c).stopQuery(cloud, c.req.param('id'), c.req.param('queryId')) + return c.json({ok: true}) + }) + }) + + app.post('/:cloud/services/datacatalog/resources/:id/tables', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).createCatalogTable(cloud, c.req.param('id'), await c.req.json()), 201)) + }) + + app.delete('/:cloud/services/datacatalog/resources/:id/tables/:tableName', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).deleteCatalogTable(cloud, c.req.param('id'), c.req.param('tableName')))) + }) + app.post('/:cloud/services/filesystems/resources/:id/mount-targets', async (c) => { const cloud = c.req.param('cloud') as CloudProvider if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) diff --git a/packages/api/src/service/CloudProxyService.ts b/packages/api/src/service/CloudProxyService.ts index 2c7289e..c3b3bd3 100644 --- a/packages/api/src/service/CloudProxyService.ts +++ b/packages/api/src/service/CloudProxyService.ts @@ -95,6 +95,11 @@ import type { DistributionInvalidation, AutoScalingCapacityInput, AccessPointInput, + CatalogTableInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, FileSystemAttachmentInput, CertificatePem, CertificateRevocationReason, @@ -966,6 +971,48 @@ export class CloudProxyService { return adapter.revokeCertificate(certificateId, reason) } + async startQuery(cloud: CloudProvider, workGroup: string, input: QueryInput): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.startQuery) throw new NotSupportedError(`Running queries is not supported for ${cloud}/query`) + return adapter.startQuery(workGroup, input) + } + + async getQueryResult(cloud: CloudProvider, workGroup: string, queryId: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.getQueryResult) throw new NotSupportedError(`Query results are not supported for ${cloud}/query`) + return adapter.getQueryResult(workGroup, queryId) + } + + async stopQuery(cloud: CloudProvider, workGroup: string, queryId: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.stopQuery) throw new NotSupportedError(`Cancelling queries is not supported for ${cloud}/query`) + return adapter.stopQuery(workGroup, queryId) + } + + async listQueries(cloud: CloudProvider, workGroup: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.listQueries) throw new NotSupportedError(`Query history is not supported for ${cloud}/query`) + return adapter.listQueries(workGroup) + } + + async describeQueryCatalog(cloud: CloudProvider, database?: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.describeQueryCatalog) throw new NotSupportedError(`Browsing the catalog is not supported for ${cloud}/query`) + return adapter.describeQueryCatalog(database) + } + + async createCatalogTable(cloud: CloudProvider, databaseName: string, input: CatalogTableInput): Promise { + const adapter = this.requireAdapter(cloud, 'datacatalog') + if (!adapter.createCatalogTable) throw new NotSupportedError(`Creating tables is not supported for ${cloud}/datacatalog`) + return adapter.createCatalogTable(databaseName, input) + } + + async deleteCatalogTable(cloud: CloudProvider, databaseName: string, tableName: string): Promise { + const adapter = this.requireAdapter(cloud, 'datacatalog') + if (!adapter.deleteCatalogTable) throw new NotSupportedError(`Deleting tables is not supported for ${cloud}/datacatalog`) + return adapter.deleteCatalogTable(databaseName, tableName) + } + async createMountTarget(cloud: CloudProvider, fileSystemId: string, subnetId: string): Promise { const adapter = this.requireAdapter(cloud, 'filesystems') if (!adapter.createMountTarget) throw new NotSupportedError(`Mount targets are not supported for ${cloud}/filesystems`) diff --git a/packages/frontend/src/api/api.ts b/packages/frontend/src/api/api.ts index 60f70ed..3ced107 100644 --- a/packages/frontend/src/api/api.ts +++ b/packages/frontend/src/api/api.ts @@ -155,6 +155,17 @@ export const apiEndpointKeys = { export: "clouds.services.certificates.export", revoke: "clouds.services.certificates.revoke", }, + query: { + catalog: "clouds.services.query.catalog", + queries: "clouds.services.query.queries", + start: "clouds.services.query.start", + result: "clouds.services.query.result", + stop: "clouds.services.query.stop", + }, + datacatalog: { + tables: "clouds.services.datacatalog.tables", + table: "clouds.services.datacatalog.table", + }, filesystems: { mountTargets: "clouds.services.filesystems.mountTargets", mountTarget: "clouds.services.filesystems.mountTarget", @@ -964,6 +975,55 @@ export const endpointRegistry: EndpointRegistry = new Map([ method: "POST", }, ], + [ + apiEndpointKeys.clouds.query.catalog, + { + path: "/clouds/:cloud/services/query/catalog", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.queries, + { + path: "/clouds/:cloud/services/query/resources/:id/queries", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.start, + { + path: "/clouds/:cloud/services/query/resources/:id/queries", + method: "POST", + }, + ], + [ + apiEndpointKeys.clouds.query.result, + { + path: "/clouds/:cloud/services/query/resources/:id/queries/:queryId", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.stop, + { + path: "/clouds/:cloud/services/query/resources/:id/queries/:queryId/stop", + method: "POST", + }, + ], + [ + apiEndpointKeys.clouds.datacatalog.tables, + { + path: "/clouds/:cloud/services/datacatalog/resources/:id/tables", + method: "POST", + }, + ], + [ + apiEndpointKeys.clouds.datacatalog.table, + { + path: "/clouds/:cloud/services/datacatalog/resources/:id/tables/:tableName", + method: "DELETE", + }, + ], [ apiEndpointKeys.clouds.filesystems.mountTargets, { diff --git a/packages/frontend/src/api/cloudProxyClient.ts b/packages/frontend/src/api/cloudProxyClient.ts index efbeab7..c3afda3 100644 --- a/packages/frontend/src/api/cloudProxyClient.ts +++ b/packages/frontend/src/api/cloudProxyClient.ts @@ -35,6 +35,11 @@ import type { DistributionInvalidation, AutoScalingCapacityInput, AccessPointInput, + CatalogTableInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, FileSystemAttachmentInput, CertificatePem, CertificateRevocationReason, @@ -1723,6 +1728,97 @@ export async function revokeCertificate( return res.data; } +export async function describeQueryCatalog( + cloud: CloudProvider, + database?: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.catalog, + requestOptions(cloud, "query", { signal, params: { database } }), + { cloud }, + ); + return res.data; +} + +export async function listQueries( + cloud: CloudProvider, + workGroup: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.queries, + requestOptions(cloud, "query", { signal }), + { cloud, id: workGroup }, + ); + return res.data; +} + +export async function startQuery( + cloud: CloudProvider, + workGroup: string, + input: QueryInput, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.start, + requestOptions(cloud, "query", { body: input }), + { cloud, id: workGroup }, + ); + return res.data; +} + +export async function getQueryResult( + cloud: CloudProvider, + workGroup: string, + queryId: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.result, + requestOptions(cloud, "query", { signal }), + { cloud, id: workGroup, queryId }, + ); + return res.data; +} + +export async function stopQuery( + cloud: CloudProvider, + workGroup: string, + queryId: string, +): Promise { + await apiClient.call( + apiEndpointKeys.clouds.query.stop, + requestOptions(cloud, "query"), + { cloud, id: workGroup, queryId }, + ); +} + +export async function createCatalogTable( + cloud: CloudProvider, + databaseName: string, + input: CatalogTableInput, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.datacatalog.tables, + requestOptions(cloud, "datacatalog", { body: input }), + { cloud, id: databaseName }, + ); + return res.data; +} + +export async function deleteCatalogTable( + cloud: CloudProvider, + databaseName: string, + tableName: string, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.datacatalog.table, + requestOptions(cloud, "datacatalog"), + { cloud, id: databaseName, tableName }, + ); + return res.data; +} + export async function createMountTarget( cloud: CloudProvider, fileSystemId: string, diff --git a/packages/frontend/src/components/DataCatalogPanel.test.tsx b/packages/frontend/src/components/DataCatalogPanel.test.tsx new file mode 100644 index 0000000..7cfa915 --- /dev/null +++ b/packages/frontend/src/components/DataCatalogPanel.test.tsx @@ -0,0 +1,92 @@ +import {QueryClient, QueryClientProvider} from '@tanstack/react-query' +import {render, screen, within} from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import {MemoryRouter} from 'react-router-dom' +import {beforeEach, expect, test, vi} from 'vitest' +import {DataCatalogPanel} from './DataCatalogPanel' +import type {CloudResource} from '@/types/resource' + +const api = vi.hoisted(() => ({ + getCloudResource: vi.fn(), + createCatalogTable: vi.fn(), + deleteCatalogTable: vi.fn(), +})) + +vi.mock('@/api/cloudProxyClient', () => api) +vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'})) + +function database(tables: unknown[] = [{ + name: 'orders', description: null, format: 'csv', location: 's3://data/orders/', delimiter: ';', headerRow: true, + columns: [{name: 'id', type: 'int'}, {name: 'amount', type: 'double'}], partitionKeys: [], createdAt: null, updatedAt: null, +}]): CloudResource { + return {id: 'shop', name: 'shop', cloud: 'aws', service: 'datacatalog', type: 'glueDatabase', region: null, createdAt: null, metadata: {description: 'Orders', tableCount: tables.length, tables}} +} + +function renderPanel(resource: CloudResource) { + api.getCloudResource.mockResolvedValue(resource) + const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) + return render() +} + +beforeEach(() => { + vi.clearAllMocks() + api.createCatalogTable.mockResolvedValue(database()) + api.deleteCatalogTable.mockResolvedValue(database([])) +}) + +test('lists tables with their format, location and columns', async () => { + renderPanel(database()) + + const row = (await screen.findByText('orders')).closest('tr') as HTMLElement + expect(within(row).getByText('s3://data/orders/')).toBeInTheDocument() + expect(within(row).getByText('semicolon-separated, header row')).toBeInTheDocument() + expect(within(row).getByText('amount')).toBeInTheDocument() + expect(screen.getByRole('link', {name: 'Query in Athena'})).toHaveAttribute('href', '/cloud-explorer/aws/query') +}) + +test('creates a CSV table from the columns entered', async () => { + const user = userEvent.setup() + renderPanel(database([])) + + const form = await screen.findByRole('form', {name: 'Create table'}) + await user.type(within(form).getByLabelText('Name'), 'events') + const location = within(form).getByLabelText('S3 location') + await user.clear(location) + await user.type(location, 's3://data/events/') + await user.type(within(form).getByLabelText('Column 1 name'), 'id') + await user.selectOptions(within(form).getByLabelText('Column 1 type'), 'int') + await user.click(within(form).getByRole('button', {name: 'Add column'})) + await user.type(within(form).getByLabelText('Column 2 name'), 'kind') + await user.click(within(form).getByRole('button', {name: 'Create table'})) + + expect(api.createCatalogTable).toHaveBeenCalledWith('aws', 'shop', { + name: 'events', location: 's3://data/events/', format: 'csv', delimiter: ',', headerRow: true, + columns: [{name: 'id', type: 'int'}, {name: 'kind', type: 'string'}], + }) + expect(await screen.findByText('Created table events. Athena can query it now.')).toBeInTheDocument() +}) + +test('JSON tables need no delimiter or header', async () => { + const user = userEvent.setup() + renderPanel(database([])) + + const form = await screen.findByRole('form', {name: 'Create table'}) + await user.selectOptions(within(form).getByLabelText('Format'), 'json') + expect(within(form).queryByLabelText('Delimiter')).not.toBeInTheDocument() + await user.type(within(form).getByLabelText('Name'), 'events') + await user.type(within(form).getByLabelText('S3 location'), 'data/events/') + await user.type(within(form).getByLabelText('Column 1 name'), 'kind') + await user.click(within(form).getByRole('button', {name: 'Create table'})) + + expect(api.createCatalogTable).toHaveBeenCalledWith('aws', 'shop', {name: 'events', location: 's3://data/events/', format: 'json', columns: [{name: 'kind', type: 'string'}]}) +}) + +test('deletes a table after confirming', async () => { + const user = userEvent.setup() + vi.spyOn(window, 'confirm').mockReturnValue(true) + renderPanel(database()) + + await user.click(await screen.findByRole('button', {name: 'Delete table orders'})) + expect(api.deleteCatalogTable).toHaveBeenCalledWith('aws', 'shop', 'orders') + expect(await screen.findByText('Deleted table orders. Its files stay in S3.')).toBeInTheDocument() +}) diff --git a/packages/frontend/src/components/DataCatalogPanel.tsx b/packages/frontend/src/components/DataCatalogPanel.tsx new file mode 100644 index 0000000..119684e --- /dev/null +++ b/packages/frontend/src/components/DataCatalogPanel.tsx @@ -0,0 +1,147 @@ +import {Plus, RefreshCw, X} from 'lucide-react' +import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' +import {useEffect, useState} from 'react' +import {Link} from 'react-router-dom' +import {createCatalogTable, deleteCatalogTable, getCloudResource} from '@/api/cloudProxyClient' +import {cloudQueryKeys} from '@/api/queries/cloudQueries' +import {useAccountId} from '@/lib/accountStore' +import {formatDateTime} from '@/lib/format' +import type {CatalogTableInput, CloudResource} from '@/types/resource' + +type Props = { + resource?: CloudResource + runtimeReachable: boolean +} + +type Column = {name: string; type: string} +type CatalogTable = { + name: string + description: string | null + format: string + location: string | null + delimiter: string | null + headerRow: boolean + columns: Column[] + partitionKeys: Column[] + updatedAt: string | null + createdAt: string | null +} +type TableForm = {name: string; location: string; format: CatalogTableInput['format']; delimiter: string; headerRow: boolean; columns: Column[]} + +/** Matches the types the API accepts; decimal(p,s) can also be typed in by hand through the API. */ +const COLUMN_TYPES = ['string', 'int', 'bigint', 'double', 'float', 'boolean', 'date', 'timestamp', 'decimal(10,2)'] +const FORMAT_LABELS: Record = {csv: 'CSV', json: 'JSON lines', parquet: 'Parquet', iceberg: 'Iceberg'} +const DELIMITER_NAMES: Record = {',': 'comma-separated', ';': 'semicolon-separated', '\t': 'tab-separated', '|': 'pipe-separated'} +const delimiterLabel = (delimiter: string | null) => DELIMITER_NAMES[delimiter ?? ','] ?? `separated by "${delimiter}"` +const emptyForm = (): TableForm => ({name: '', location: 's3://', format: 'csv', delimiter: ',', headerRow: true, columns: [{name: '', type: 'string'}]}) + +export function DataCatalogPanel({resource, runtimeReachable}: Props) { + const accountId = useAccountId() + const queryClient = useQueryClient() + const [form, setForm] = useState(emptyForm) + const [notice, setNotice] = useState('') + const detailKey = ['glue-database', accountId, resource?.id] + + const detailQuery = useQuery({ + queryKey: detailKey, + queryFn: ({signal}) => getCloudResource('aws', 'datacatalog', resource?.id ?? '', signal), + enabled: Boolean(resource) && runtimeReachable, + }) + + useEffect(() => { + setNotice('') + setForm(emptyForm()) + }, [resource?.id]) + + const settle = (updated: CloudResource, message: string) => { + queryClient.setQueryData(detailKey, updated) + setNotice(message) + return queryClient.invalidateQueries({queryKey: cloudQueryKeys.resources('aws', 'datacatalog')}) + } + const createTable = useMutation({ + mutationFn: () => createCatalogTable('aws', resource?.id ?? '', { + name: form.name.trim(), + location: form.location.trim(), + format: form.format, + ...(form.format === 'csv' ? {delimiter: form.delimiter, headerRow: form.headerRow} : {}), + columns: form.columns.map((column) => ({name: column.name.trim(), type: column.type})), + }), + onSuccess: (updated) => { + const name = form.name.trim() + setForm(emptyForm()) + return settle(updated, `Created table ${name}. Athena can query it now.`) + }, + }) + const deleteTable = useMutation({ + mutationFn: (tableName: string) => deleteCatalogTable('aws', resource?.id ?? '', tableName), + onSuccess: (updated, tableName) => settle(updated, `Deleted table ${tableName}. Its files stay in S3.`), + }) + + if (!resource) { + return

Select a database

Choose a database to see its tables, or describe files in S3 as a new table.

+ } + + const database = detailQuery.data ?? resource + const tables = Array.isArray(database.metadata.tables) ? database.metadata.tables as CatalogTable[] : [] + const disabled = !runtimeReachable || createTable.isPending || deleteTable.isPending + const error = detailQuery.error ?? createTable.error ?? deleteTable.error + const setColumn = (index: number, patch: Partial) => setForm({...form, columns: form.columns.map((column, i) => i === index ? {...column, ...patch} : column)}) + + return
+
+

Glue database

{database.metadata.description ? String(database.metadata.description) : 'Tables here describe files in S3 so Athena can query them.'}
+
+ {tables.length > 0 && Query in Athena} + +
+
+ +

Tables

{tables.length === 1 ? '1 table' : `${tables.length} tables`}
+ {tables.length === 0 ?

No tables yet. Describe a folder of CSV, JSON or Parquet files below.

+ :
{tables.map((table) => ( + + + + + + + + + ))}
TableFormatLocationColumnsUpdated
{table.name}{table.description && <>
{table.description}}		{FORMAT_LABELS[table.format] ?? table.format}{table.format === 'csv' && <>
{delimiterLabel(table.delimiter)}{table.headerRow ? ', header row' : ''}}		{table.location ?? '—'}{table.columns.map((column) => {column.name} {column.type})} + {table.partitionKeys.length > 0 && partitioned by {table.partitionKeys.map((key) => key.name).join(', ')}}{formatDateTime(table.updatedAt ?? table.createdAt) ?? '—'}
} + +
{ event.preventDefault(); setNotice(''); createTable.mutate() }}> +

New table

+
+ + + + {form.format === 'csv' && <> + + + } +
+
+ Columns + {form.columns.map((column, index) =>
+ setColumn(index, {name: event.target.value})} required pattern="[a-z_][a-z0-9_]*" title="Lowercase letters, numbers and underscores"/> + + +
)} + +
+

List the columns in the order they appear in the files. For JSON, use the field names.

+
+
+ + {notice &&

{notice}

} + {error instanceof Error &&

{error.message}

} +
+} diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index fb8926c..3e6655f 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -70,6 +70,8 @@ import { Route53RecordsPanel } from "@/components/Route53RecordsPanel"; import { CloudFrontPanel } from "@/components/CloudFrontPanel"; import { AutoScalingPanel } from "@/components/AutoScalingPanel"; import { FileSystemPanel } from "@/components/FileSystemPanel"; +import { DataCatalogPanel } from "@/components/DataCatalogPanel"; +import { QueryPanel } from "@/components/QueryPanel"; import { CertificatePanel } from "@/components/CertificatePanel"; import { PipePanel } from "@/components/PipePanel"; import { CloudWatchAlarmPanel } from "@/components/CloudWatchAlarmPanel"; @@ -1003,6 +1005,12 @@ export function DynamicResourceView({ {service === "filesystems" && cloud === "aws" && ( )} + {service === "datacatalog" && cloud === "aws" && ( + + )} + {service === "query" && cloud === "aws" && ( + + )} {service === "cdn" && cloud === "aws" && ( )} diff --git a/packages/frontend/src/components/QueryPanel.test.tsx b/packages/frontend/src/components/QueryPanel.test.tsx new file mode 100644 index 0000000..b650f3e --- /dev/null +++ b/packages/frontend/src/components/QueryPanel.test.tsx @@ -0,0 +1,100 @@ +import {QueryClient, QueryClientProvider} from '@tanstack/react-query' +import {render, screen, within} from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import {MemoryRouter} from 'react-router-dom' +import {beforeEach, expect, test, vi} from 'vitest' +import {QueryPanel} from './QueryPanel' +import type {CloudResource, QueryExecutionSummary} from '@/types/resource' + +const api = vi.hoisted(() => ({ + describeQueryCatalog: vi.fn(), + getCloudResource: vi.fn(), + listQueries: vi.fn(), + startQuery: vi.fn(), + getQueryResult: vi.fn(), + stopQuery: vi.fn(), + storageObjectDownloadUrl: vi.fn((_cloud: string, bucket: string, key: string) => `/download/${bucket}/${key}`), +})) + +vi.mock('@/api/cloudProxyClient', () => api) +vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'})) + +const workGroup: CloudResource = {id: 'analysts', name: 'analysts', cloud: 'aws', service: 'query', type: 'athenaWorkGroup', region: null, createdAt: null, status: 'ENABLED', metadata: {outputLocation: 's3://results/'}} + +function execution(overrides: Partial = {}): QueryExecutionSummary { + return { + id: 'q1', sql: 'SELECT * FROM orders LIMIT 10', database: 'shop', workGroup: 'analysts', state: 'SUCCEEDED', stateReason: null, statementType: 'DML', + submittedAt: '2026-10-25T21:00:00.000Z', completedAt: null, runtimeMs: 42, dataScannedBytes: 0, outputLocation: 's3://results/q1.csv', ...overrides, + } +} + +function renderPanel() { + const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) + return render() +} + +beforeEach(() => { + vi.clearAllMocks() + api.getCloudResource.mockResolvedValue({...workGroup, metadata: {outputLocation: 's3://results/athena/'}}) + api.describeQueryCatalog.mockResolvedValue({databases: ['logs', 'shop'], database: 'logs', tables: [{name: 'orders', columns: [{name: 'id', type: 'int'}, {name: 'amount', type: 'double'}]}]}) + api.listQueries.mockResolvedValue([]) + api.startQuery.mockResolvedValue(execution({state: 'RUNNING'})) + api.getQueryResult.mockResolvedValue({execution: execution(), columns: [{name: 'id', type: 'integer'}, {name: 'amount', type: 'double'}], rows: [['1', '10.5'], ['2', null]], truncated: false}) +}) + +test('starts from a preview of the first table and lists its columns', async () => { + renderPanel() + + expect(await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10')).toBeInTheDocument() + expect(screen.getByText('amount')).toBeInTheDocument() + expect(screen.getByLabelText('Database')).toHaveValue('logs') + expect(await screen.findByText('s3://results/athena/')).toBeInTheDocument() +}) + +test('runs the query in the workgroup and shows its rows', async () => { + const user = userEvent.setup() + renderPanel() + + await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10') + await user.click(screen.getByRole('button', {name: 'Run'})) + + expect(api.startQuery).toHaveBeenCalledWith('aws', 'analysts', {sql: 'SELECT * FROM orders LIMIT 10', database: 'logs'}) + expect(await screen.findByText('10.5')).toBeInTheDocument() + expect(screen.getByText('NULL')).toBeInTheDocument() + expect(screen.getByText('2 rows')).toBeInTheDocument() + expect(screen.getByRole('link', {name: 'Download CSV'})).toHaveAttribute('href', '/download/results/q1.csv') +}) + +test('shows why a query failed', async () => { + const user = userEvent.setup() + api.getQueryResult.mockResolvedValue({execution: execution({state: 'FAILED', stateReason: 'Table nope does not exist'}), columns: [], rows: [], truncated: false}) + renderPanel() + + const sql = await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10') + await user.clear(sql) + await user.type(sql, 'SELECT * FROM nope') + await user.keyboard('{Control>}{Enter}{/Control}') + + expect(api.startQuery).toHaveBeenCalledWith('aws', 'analysts', {sql: 'SELECT * FROM nope', database: 'logs'}) + expect(await screen.findByText('Table nope does not exist')).toBeInTheDocument() + expect(screen.queryByRole('link', {name: 'Download CSV'})).not.toBeInTheDocument() +}) + +test('opens a query from the history', async () => { + const user = userEvent.setup() + api.listQueries.mockResolvedValue([execution({id: 'q0', sql: 'SELECT count(*) FROM orders', database: 'shop'})]) + renderPanel() + + await user.click(await screen.findByRole('button', {name: 'SELECT count(*) FROM orders'})) + expect(screen.getByLabelText('SQL')).toHaveValue('SELECT count(*) FROM orders') + expect(api.getQueryResult).toHaveBeenCalledWith('aws', 'analysts', 'q0', expect.anything()) +}) + +test('points to the Glue Data Catalog when there is nothing to query', async () => { + api.describeQueryCatalog.mockResolvedValue({databases: [], database: null, tables: []}) + renderPanel() + + const link = await screen.findByRole('link', {name: 'Glue Data Catalog'}) + expect(link).toHaveAttribute('href', '/cloud-explorer/aws/datacatalog') + expect(within(screen.getByRole('form', {name: 'Query'})).getByRole('button', {name: 'Run'})).toBeDisabled() +}) diff --git a/packages/frontend/src/components/QueryPanel.tsx b/packages/frontend/src/components/QueryPanel.tsx new file mode 100644 index 0000000..84a4c81 --- /dev/null +++ b/packages/frontend/src/components/QueryPanel.tsx @@ -0,0 +1,185 @@ +import {Download, Play, RefreshCw, Square} from 'lucide-react' +import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' +import {useEffect, useMemo, useState} from 'react' +import {Link} from 'react-router-dom' +import {describeQueryCatalog, getCloudResource, getQueryResult, listQueries, startQuery, stopQuery, storageObjectDownloadUrl} from '@/api/cloudProxyClient' +import {useAccountId} from '@/lib/accountStore' +import {formatDateTime} from '@/lib/format' +import type {CloudResource, QueryExecutionSummary} from '@/types/resource' + +type Props = { + resource?: CloudResource + runtimeReachable: boolean +} + +/** Athena reports a running query's state; it is re-read this often until it finishes. */ +const RUNNING_POLL_MS = 1000 +const FINISHED = new Set(['SUCCEEDED', 'FAILED', 'CANCELLED']) + +export function QueryPanel({resource, runtimeReachable}: Props) { + const accountId = useAccountId() + const queryClient = useQueryClient() + const workGroup = resource?.id ?? '' + const [database, setDatabase] = useState('') + const [sql, setSql] = useState('') + const [activeQueryId, setActiveQueryId] = useState(null) + const [seededFor, setSeededFor] = useState(null) + const historyKey = useMemo(() => ['athena-history', accountId, workGroup], [accountId, workGroup]) + + const workGroupQuery = useQuery({ + queryKey: ['athena-workgroup', accountId, workGroup], + queryFn: ({signal}) => getCloudResource('aws', 'query', workGroup, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const catalogQuery = useQuery({ + queryKey: ['athena-catalog', accountId, database], + queryFn: ({signal}) => describeQueryCatalog('aws', database || undefined, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const historyQuery = useQuery({ + queryKey: historyKey, + queryFn: ({signal}) => listQueries('aws', workGroup, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const resultQuery = useQuery({ + queryKey: ['athena-result', accountId, workGroup, activeQueryId], + queryFn: ({signal}) => getQueryResult('aws', workGroup, activeQueryId ?? '', signal), + enabled: Boolean(activeQueryId) && runtimeReachable, + refetchInterval: (query) => query.state.data && !FINISHED.has(query.state.data.execution.state) ? RUNNING_POLL_MS : false, + }) + + const catalog = catalogQuery.data + const chosenDatabase = database || catalog?.database || '' + const firstTable = catalog?.tables[0]?.name + useEffect(() => { + setActiveQueryId(null) + setSql('') + }, [workGroup]) + // Start each workgroup from a query that runs as-is, the way the Athena console opens on a table. + useEffect(() => { + if (!firstTable || seededFor === workGroup) return + setSeededFor(workGroup) + setSql((current) => current || `SELECT * FROM ${quote(firstTable)} LIMIT 10`) + }, [firstTable, seededFor, workGroup]) + + const execution = resultQuery.data?.execution + const finished = execution ? FINISHED.has(execution.state) : true + const executionState = execution?.state + useEffect(() => { + if (executionState && FINISHED.has(executionState)) void queryClient.invalidateQueries({queryKey: historyKey}) + }, [executionState, historyKey, queryClient]) + + const run = useMutation({ + mutationFn: () => startQuery('aws', workGroup, {sql, ...(chosenDatabase ? {database: chosenDatabase} : {})}), + onSuccess: (started) => { + setActiveQueryId(started.id) + return queryClient.invalidateQueries({queryKey: historyKey}) + }, + }) + const cancel = useMutation({ + mutationFn: () => stopQuery('aws', workGroup, activeQueryId ?? ''), + onSuccess: () => resultQuery.refetch(), + }) + + if (!resource) { + return

Select a workgroup

Choose a workgroup to run SQL against the tables in the Glue Data Catalog.

+ } + + const outputLocation = (workGroupQuery.data ?? resource).metadata.outputLocation + const disabled = !runtimeReachable || run.isPending + const result = resultQuery.data + const error = run.error ?? cancel.error ?? resultQuery.error ?? catalogQuery.error + const history = historyQuery.data ?? [] + const download = execution?.outputLocation ? s3Location(execution.outputLocation) : null + const submit = () => { if (sql.trim() && !disabled) run.mutate() } + const openHistory = (entry: QueryExecutionSummary) => { + setActiveQueryId(entry.id) + setSql(entry.sql) + if (entry.database) setDatabase(entry.database) + } + + return
+
+

Athena query editor

Workgroup {resource.name}{outputLocation ? <> · results in {String(outputLocation)} : null}
+
+ +
+
+ +
+ + +
+
{ event.preventDefault(); submit() }}> +