diff --git a/docsfera.json b/docsfera.json index da23ed4..2653fe1 100644 --- a/docsfera.json +++ b/docsfera.json @@ -5206,6 +5206,44 @@ } }, "/platform/pipelines/{pipelineId}/inputs/{inputId}/tables": { + "delete": { + "operationId": "PlatformApiController_deleteTables", + "summary": "Batch-remove tables from an input; reconfigures the CDC connectors once", + "parameters": [ + { + "name": "pipelineId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + }, + { + "name": "inputId", + "required": true, + "in": "path", + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "" + } + }, + "tags": [ + "Platform API" + ], + "security": [ + { + "access-token": [] + }, + { + "access-token": [] + } + ] + }, "post": { "operationId": "PlatformApiController_addTable", "summary": "Add a CDC table to an input and dispatch its platform jobs", @@ -10887,6 +10925,25 @@ "updated_at" ] }, + "CdcColumnReq": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "type": { + "type": "string" + }, + "is_primary_key": { + "type": "boolean" + } + }, + "required": [ + "name", + "type", + "is_primary_key" + ] + }, "CdcTableReq": { "type": "object", "properties": { @@ -10901,12 +10958,40 @@ "items": { "type": "string" } + }, + "iceberg_table_name": { + "type": "string" + }, + "columns": { + "type": "array", + "items": { + "$ref": "#/components/schemas/CdcColumnReq" + } } }, "required": [ "name" ] }, + "IcebergDestinationReq": { + "type": "object", + "properties": { + "namespace": { + "type": "string" + } + }, + "required": [ + "namespace" + ] + }, + "CdcDestinationReq": { + "type": "object", + "properties": { + "iceberg": { + "$ref": "#/components/schemas/IcebergDestinationReq" + } + } + }, "CreateCdcInputReq": { "type": "object", "properties": { @@ -10924,6 +11009,9 @@ }, "read_only": { "type": "boolean" + }, + "destination": { + "$ref": "#/components/schemas/CdcDestinationReq" } }, "required": [ diff --git a/package-lock.json b/package-lock.json index 1ae57c4..3b8fcd3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,7 +16,7 @@ "@aws-sdk/lib-dynamodb": "^3.414.0", "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", - "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.0.tgz", + "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.1.tgz", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", @@ -1735,9 +1735,9 @@ } }, "node_modules/@dadosfera/protospack-v2": { - "version": "3.41.0-cdc-iceberg.0", - "resolved": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.0.tgz", - "integrity": "sha512-qPs0LukP5BJnybl3JQqqbYQ9oi5SJ86+4A3zd9zisrqqzFHv3o9c0CSwcimwyT3CWm/147vq5LMQy0ulal0Ung==", + "version": "3.41.0-cdc-iceberg.1", + "resolved": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.1.tgz", + "integrity": "sha512-9skEKTRbsgcnACfk6YDg2i9IePdS2FTgH5ngacaxljVv+4+oHsxcWRkIsTbvf/dC6RDAdjd38cU+oa9VGOTCEg==", "license": "ISC", "dependencies": { "@grpc/grpc-js": "^1.9.3", diff --git a/package.json b/package.json index 17c03bf..faf536f 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,7 @@ "@aws-sdk/lib-dynamodb": "^3.414.0", "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", - "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.0.tgz", + "@dadosfera/protospack-v2": "file:../protospack-v2/dadosfera-protospack-v2-3.41.0-cdc-iceberg.1.tgz", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", diff --git a/src/modules/inputs/dtos/input.model.ts b/src/modules/inputs/dtos/input.model.ts index 2643521..5f7e330 100644 --- a/src/modules/inputs/dtos/input.model.ts +++ b/src/modules/inputs/dtos/input.model.ts @@ -66,6 +66,15 @@ export class CreateInputReq extends OmitType(Input, [ 'updated_at', ]) {} +export class CdcColumnReq { + @ApiProperty() + name: string; + @ApiProperty() + type: string; + @ApiProperty() + is_primary_key: boolean; +} + export class CdcTableReq { @ApiProperty() name: string; @@ -76,6 +85,8 @@ export class CdcTableReq { // Advisory/reserved: the platform currently derives the Iceberg table name itself (create_iceberg_table_name); this value is persisted but not yet consumed on the create/add path. Do not treat as the authoritative table name. @ApiPropertyOptional() iceberg_table_name?: string; + @ApiPropertyOptional({ type: [CdcColumnReq] }) + columns?: CdcColumnReq[]; } export class IcebergDestinationReq { diff --git a/src/modules/inputs/inputs.service.spec.ts b/src/modules/inputs/inputs.service.spec.ts index fed634d..46d9e38 100644 --- a/src/modules/inputs/inputs.service.spec.ts +++ b/src/modules/inputs/inputs.service.spec.ts @@ -58,6 +58,35 @@ describe('InputsService.createCdc', () => { ); }); + it('forwards per-table columns to the gRPC request', async () => { + const body: CreateCdcInputReq = { + name: 'CDC Columns Test', + plugin: 'mysql_cdc', + read_only: true, + tables: [ + { + name: 'orders', + table_schema: 'mydb', + primary_keys: ['id'], + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], + }, + ], + }; + + await service.createCdc({ body, info }); + + expect(inputCreateCdcMock).toHaveBeenCalledTimes(1); + const sentRequest = inputCreateCdcMock.mock.calls[0][0]; + + expect(sentRequest.input.tables[0].columns).toEqual([ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ]); + }); + it('back-compat: a body with no destination sends destination undefined, not an error', async () => { const body: CreateCdcInputReq = { name: 'CDC Legacy Test', diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index c221520..38b3a4b 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -198,6 +198,7 @@ export class InputsService { name: t.name, // canonical identity == table_name (in-factory also backfills) primary_keys: t.primary_keys ?? [], iceberg_table_name: t.iceberg_table_name, + columns: t.columns ?? [], })), destination: body.destination, }, diff --git a/src/modules/platform-api/platform-api.controller.spec.ts b/src/modules/platform-api/platform-api.controller.spec.ts index 57127ca..fd6cf78 100644 --- a/src/modules/platform-api/platform-api.controller.spec.ts +++ b/src/modules/platform-api/platform-api.controller.spec.ts @@ -163,6 +163,7 @@ describe('PlatformApiController - addTable', () => { table_name: 'orders', primary_keys: ['id'], name: 'orders', + columns: [], }, info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, }); @@ -201,6 +202,37 @@ describe('PlatformApiController - addTable', () => { primary_keys: ['id'], name: 'orders', iceberg_table_name: 'cdc_raw.public__orders', + columns: [], + }, + info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, + }); + }); + + it('addTable: carries columns on the added table through to AddCdcTable', async () => { + inputsService.addCdcTable.mockResolvedValue({ input: {} }); + pipelinesClientService.addCdcJobs.mockResolvedValue({ job_ids: ['p_4'], skipped: [] }); + + const columnsBody = { + ...body, + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], + }; + + await controller.addTable('pid', 'iid', columnsBody, mockUser); + + expect(inputsService.addCdcTable).toHaveBeenCalledWith({ + id: 'iid', + table: { + table_schema: 'public', + table_name: 'orders', + primary_keys: ['id'], + name: 'orders', + columns: [ + { name: 'id', type: 'int', is_primary_key: true }, + { name: 'descr', type: 'varchar(255)', is_primary_key: false }, + ], }, info: { customer_id: 'c1', customer: 'cust', user_id: 'u1' }, }); diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts index 8355ffa..cbb9aef 100644 --- a/src/modules/platform-api/platform-api.controller.ts +++ b/src/modules/platform-api/platform-api.controller.ts @@ -1079,6 +1079,8 @@ export class PlatformApiController { // Iceberg destination only (protospack CdcTable.iceberg_table_name); // absent for snowflake, back-compat. iceberg_table_name?: string; + // Source column schema for iceberg deduped table pre-create (protospack CdcTable.columns). + columns?: { name: string; type: string; is_primary_key: boolean }[]; }, @User() user: RequestUser, ) { @@ -1094,6 +1096,7 @@ export class PlatformApiController { primary_keys: body.primary_keys, name: body.table_name, iceberg_table_name: body.iceberg_table_name, + columns: body.columns ?? [], }; this.logger.info('addTable: appending CDC table to DynamoDB', { inputId, tableName: body.table_name });