Compare commits

..
2 changed files with 71 additions and 13 deletions
@@ -83,7 +83,7 @@ describe('ConnectionTestService catalog cache', () => {
});
});
it('maps cached columns to the existing table metadata contract', async () => {
it('maps cached columns and derives references from the engine allowlist', async () => {
connectionsApiService.proxy.mockResolvedValue({
columns: [
{
@@ -91,6 +91,16 @@ describe('ConnectionTestService catalog cache', () => {
data_type: 'bigint',
is_primary_key: true,
},
{
column_name: 'name',
data_type: 'varchar',
is_primary_key: false,
},
],
});
platformApiService.proxy.mockResolvedValue({
allowed_datatypes: [
{ engine: 'postgresql', allowed_datatypes: ['bigint', 'timestamp'] },
],
});
@@ -110,13 +120,10 @@ describe('ConnectionTestService catalog cache', () => {
{
table_name: 'customers',
columns: [
{
name: 'id',
type: 'bigint',
is_primary_key: true,
},
{ name: 'id', type: 'bigint', is_primary_key: true },
{ name: 'name', type: 'varchar', is_primary_key: false },
],
references: [],
references: [{ name: 'id', type: 'bigint', is_primary_key: true }],
},
],
});
@@ -125,6 +132,39 @@ describe('ConnectionTestService catalog cache', () => {
'/connection_catalog/config-id/schemas/public/tables/customers/columns',
user,
);
expect(platformApiService.proxy).toHaveBeenCalledWith(
'GET',
'/jobs/jdbc/configs/allowed_datatypes',
user,
);
});
it('returns empty references when the allowlist call fails', async () => {
connectionsApiService.proxy.mockResolvedValue({
columns: [{ column_name: 'id', data_type: 'bigint', is_primary_key: true }],
});
platformApiService.proxy.mockRejectedValue(new Error('platform down'));
await expect(
service.getTableMetadata(
{
connection_id: 'config-id',
plugin: 'postgresql',
schema: 'public',
table_list: ['customers'],
},
user,
),
).resolves.toEqual({
operation_result: true,
tables_metadata: [
{
table_name: 'customers',
columns: [{ name: 'id', type: 'bigint', is_primary_key: true }],
references: [],
},
],
});
});
it('submits a catalog refresh without holding the request open', async () => {
@@ -213,6 +213,20 @@ export class ConnectionTestService {
body: GetTableMetadataReq,
user: RequestUser,
): Promise<GetTableMetadataRes> {
// Columns eligible as the incremental reference field are the ones whose
// data type is allowed for this engine (e.g. int/date/timestamp). The
// allowlist is owned by the platform API, keyed by engine === plugin.
const allowedByEngine = await this.platformApiService
.proxy('GET', '/jobs/jdbc/configs/allowed_datatypes', user)
.catch(() => null);
const allowedDataTypes: string[] =
allowedByEngine?.allowed_datatypes?.find(
(datatypes) => datatypes.engine === body.plugin,
)?.allowed_datatypes ?? [];
const allowedSet = new Set(
allowedDataTypes.map((type) => type.toLowerCase()),
);
const tables_metadata = await Promise.all(
body.table_list.map(async (table_name) => {
const result = await this.connectionsApiService.proxy(
@@ -222,14 +236,18 @@ export class ConnectionTestService {
`/tables/${encodeURIComponent(table_name)}/columns`,
user,
);
const columns = result.columns.map((column) => ({
name: column.column_name,
type: column.data_type,
is_primary_key: column.is_primary_key,
}));
const references = columns.filter((column) =>
allowedSet.has(String(column.type).toLowerCase()),
);
return {
table_name,
columns: result.columns.map((column) => ({
name: column.column_name,
type: column.data_type,
is_primary_key: column.is_primary_key,
})),
references: [],
columns,
references,
};
}),
);