FIX: enhance getTableMetadata to filter columns by allowed data types

FIX: CDC live status decodes int64 offsets as numbers

src/modules/connection-test/connection-test.service.ts

@@ -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,
         };
       }),
     );

src/modules/inputs/cdc-table.mapper.ts

@@ -3,11 +3,15 @@ import {
   CdcTable,
 } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities';
 
-/** What a caller knows about a CDC table before it is stored. */
+/** What a caller knows about a CDC table before it is stored. The source
+ *  table is named `table_name` on the platform-facing bodies and `name` on
+ *  the create DTO (CdcTableReq); either works — deriving one from the other
+ *  happens only here. */
 export interface CdcTableInput {
   // Optional on the create DTO (CdcTableReq); the platform validates it.
   table_schema?: string;
-  table_name: string;
+  table_name?: string;
+  name?: string;
   primary_keys?: string[];
   iceberg_table_name?: string;
   iceberg_qualify_table_name?: string;
@@ -25,10 +29,11 @@ export interface CdcTableInput {
  * `name ?? table_name`).
  */
 export function toCdcTable(table: CdcTableInput): CdcTable {
+  const tableName = table.table_name ?? table.name;
   return {
     table_schema: table.table_schema,
-    table_name: table.table_name,
-    name: table.table_name,
+    table_name: tableName,
+    name: tableName,
     primary_keys: table.primary_keys ?? [],
     iceberg_table_name: table.iceberg_table_name,
     iceberg_qualify_table_name: table.iceberg_qualify_table_name,

src/modules/inputs/inputs.service.ts

@@ -196,9 +196,7 @@ export class InputsService {
         name: body.name,
         plugin: body.plugin,
         read_only: body.read_only ?? true,
-        tables: body.tables.map((t) =>
-          toCdcTable({ ...t, table_name: t.name }),
-        ),
+        tables: body.tables.map(toCdcTable),
         destination: body.destination,
       },
       info,

src/modules/platform-api/pipeline-tables.service.ts

@@ -30,7 +30,7 @@ interface PlatformPipeline {
  */
 @Injectable()
 export class PipelineTablesService {
-  private logger: DadosferaLogger['logger'];
+  logger: DadosferaLogger;
 
   constructor(
     private readonly platformApiService: PlatformApiService,