diff --git a/src/modules/platform-api/platform-api.controller.ts b/src/modules/platform-api/platform-api.controller.ts
index 383647c..78b273e 100644
--- a/src/modules/platform-api/platform-api.controller.ts
+++ b/src/modules/platform-api/platform-api.controller.ts
@@ -22,7 +22,7 @@ import { User, RequestUser } from '../../decorators/user.decorator';
 import { PlatformApiService } from './platform-api.service';
 import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
 import { ElasticsearchService } from '../../services/elasticsearch';
-import { DynamoDBService } from '../../services/dynamodb';
+import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
 import { CustomersService } from '../customers/customers.service';
 import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
 
@@ -155,7 +155,7 @@ export class PlatformApiController {
    * Extract and transform tables from jobs for DynamoDB input.
    * Maps connector-specific fields to a common table format.
    *
-   * - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name (reference_column)
+   * - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name/type (reference_column object)
    * - Singer: type maps replication_method (FULL_TABLE -> full_load, INCREMENTAL -> incremental), no columns
    * - S3: same mapping as Singer, no columns
    */
@@ -163,7 +163,7 @@ export class PlatformApiController {
     name: string;
     type: string;
     columns?: string[];
-    reference_column?: string;
+    reference_column?: ReferenceColumn;
   }> {
     if (!jobs || jobs.length === 0) return [];
 
@@ -171,7 +171,7 @@ export class PlatformApiController {
       name: string;
       type: string;
       columns?: string[];
-      reference_column?: string;
+      reference_column?: ReferenceColumn;
     }> = [];
 
     for (const job of jobs) {
@@ -179,12 +179,12 @@ export class PlatformApiController {
       if (!input) continue;
 
       if (connector === 'jdbc') {
-        // JDBC: table_name, load_type, column_include_list, incremental_column_name
+        // JDBC: table_name, load_type, column_include_list, incremental_column_name/type
         const table: {
           name: string;
           type: string;
           columns?: string[];
-          reference_column?: string;
+          reference_column?: ReferenceColumn;
         } = {
           name: input.table_name || '',
           type: input.load_type || 'full_load',
@@ -193,8 +193,11 @@ export class PlatformApiController {
           table.columns = input.column_include_list;
         }
         if (input.incremental_column_name) {
-          // reference_column is stored as a string (column name)
-          table.reference_column = input.incremental_column_name;
+          // reference_column is stored as an object with name and type
+          table.reference_column = {
+            name: input.incremental_column_name,
+            type: input.incremental_column_type || 'unknown',
+          };
         }
         tables.push(table);
       } else if (connector === 'singer' || connector === 's3') {
@@ -317,11 +320,11 @@ export class PlatformApiController {
       }
 
       // Build changes for DynamoDB table entry
-      // reference_column is stored as a string (column name), not an object
+      // reference_column is stored as an object with name and type
       const changes: {
         type?: string;
         columns?: string[];
-        reference_column?: string | null;
+        reference_column?: ReferenceColumn | null;
       } = {};
 
 
@@ -332,8 +335,15 @@ export class PlatformApiController {
         changes.columns = body.column_include_list;
       }
       if ('incremental_column_name' in body) {
-        // reference_column is just the column name as a string
-        changes.reference_column = body.incremental_column_name || null;
+        // reference_column is stored as an object with name and type
+        if (body.incremental_column_name) {
+          changes.reference_column = {
+            name: body.incremental_column_name,
+            type: body.incremental_column_type || 'unknown',
+          };
+        } else {
+          changes.reference_column = null;
+        }
       }
 
       // Update DynamoDB if there are changes
@@ -466,11 +476,13 @@ export class PlatformApiController {
         });
       }
 
+      const pipelineType = this.mapConnectorToDynamoType(connectorType);
       this.logger.info('Syncing pipeline to Elasticsearch', {
         customerName: user.customer_name,
         pipelineId: body.id,
         plugin,
         connector: connectorType,
+        type: pipelineType,
         properties,
         inputId,
       });
@@ -490,7 +502,7 @@ export class PlatformApiController {
           cron: body.cron,
           tables: inputId,
           properties,
-          type: this.mapConnectorToDynamoType(connectorType),
+          type: pipelineType,
         },
         connector,
       );
diff --git a/src/services/dynamodb/dynamodb.service.ts b/src/services/dynamodb/dynamodb.service.ts
index 099eb99..15dde41 100644
--- a/src/services/dynamodb/dynamodb.service.ts
+++ b/src/services/dynamodb/dynamodb.service.ts
@@ -11,6 +11,11 @@ import { DadosferaLogger } from '@dadosfera/dadosfera-logs';
 import { v4 as uuid } from 'uuid';
 import { DYNAMODB_CONFIG } from './dynamodb.config';
 
+export interface ReferenceColumn {
+  name: string;
+  type: string;
+}
+
 export interface InputDocument {
   id: string;
   client_id: string;
@@ -24,7 +29,7 @@ export interface InputDocument {
     name: string;
     type: string;
     columns?: string[];
-    reference_column?: string;
+    reference_column?: ReferenceColumn;
   }>;
   credentials?: Record<string, any>;
 }
@@ -62,7 +67,7 @@ export class DynamoDBService {
         name: string;
         type: string;
         columns?: string[];
-        reference_column?: string;
+        reference_column?: ReferenceColumn;
       }>;
     },
   ): Promise<InputDocument> {
@@ -157,7 +162,7 @@ export class DynamoDBService {
     changes: {
       type?: string;
       columns?: string[];
-      reference_column?: string | null;
+      reference_column?: ReferenceColumn | null;
     },
   ): Promise<void> {
     const dynamoTableName = DYNAMODB_CONFIG.inputsTable();