feat(api): add the Glue Data Catalog to the cloud explorer
CI / checks (pull_request) Successful in 28m15s

Create and delete Glue databases, and describe a folder of CSV, JSON-lines or
Parquet files in S3 as a table: its columns, and for CSV the delimiter and
whether the first line is a header. Tables are written with the storage
settings Athena expects, so TheCloud's Athena can query one as soon as it
exists. Glue jobs and crawlers are not offered: the runtime stores them but
never runs them.
This commit is contained in:
Zakaria
2026-10-25 17:51:16 -04:00
parent 26b635db0b
commit 69d80c60a3
23 changed files with 918 additions and 1 deletions
@@ -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<string, unknown>
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<Record<string, unknown>>
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'})
})
})
@@ -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<CloudResource[]> {
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<CloudResource | null> {
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<CloudResource> {
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<void> {
await this.glue.send(new DeleteDatabaseCommand({Name: id}))
}
async createCatalogTable(databaseName: string, input: CatalogTableInput): Promise<CloudResource> {
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<CloudResource> {
await this.glue.send(new DeleteTableCommand({DatabaseName: databaseName, Name: tableName}))
return this.detail(databaseName)
}
private async database(name: string): Promise<Database | null> {
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<CloudResource> {
const detail = await this.get(name)
if (!detail) throw new NotFoundError(`Database ${name} not found`)
return detail
}
private async tables(databaseName: string): Promise<Table[]> {
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<string>()
return value.map((column, index) => {
const entry = typeof column === 'object' && column !== null ? column as Record<string, unknown> : {}
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
}