Remove redundant task_execution_id column (#31)

CLAUDE.md

@@ -397,32 +397,32 @@ task.completed event: taskExecutionId = 97469968-64b5-3cdd-b809-e58e4b6e3be7
 - ON CONFLICT: `(task_execution_id)` → never matches
 - Result: Always INSERTs new row (4 rows for 2 tasks)
 
-**Solution (V2 Migration):**
+**Current Solution:**
 - Primary key: `(instance_id, task_position)` - composite key
 - ON CONFLICT: `(instance_id, task_position)` → matches on same task
 - Result: UPDATEs existing row on subsequent events (2 rows for 2 tasks)
+- **No task_execution_id column** - ID is derived from composite key
 
 **Rationale:**
 - `(instance_id, task_position)` uniquely identifies a task execution within a workflow
 - Same workflow + same position = same task (regardless of changing taskExecutionId)
 - Enables proper UPSERT behavior despite Quarkus Flow's ID generation
+- TaskExecution.id is derived as `instanceId + ":" + taskPosition`
 
-**Migration:** `V2__fix_task_instances_composite_key.sql`
-- Drops old PK on `task_execution_id`
-- Adds composite PK on `(instance_id, task_position)`
-- Updates trigger function to use composite key for conflict resolution
-- Cleans up existing duplicates (keeps latest by `last_event_time`)
-- Adds index on `task_execution_id` for GraphQL queries by ID
+**Schema (V1):**
+- Composite PK on `(instance_id, task_position)`
+- Trigger function uses composite key for conflict resolution
+- No task_execution_id column (derived in application layer)
 
 **DO:**
 - ✅ Use composite key `(instance_id, task_position)` for task identity
-- ✅ Keep `task_execution_id` as a regular column (still exposed in GraphQL)
+- ✅ Derive TaskExecution.id from `instanceId + ":" + taskPosition`
 - ✅ Trust `task_position` to be stable across task lifecycle
 
 **DON'T:**
-- ❌ Don't rely on `taskExecutionId` for UPSERT logic (it changes per event)
+- ❌ Don't add task_execution_id column to database
+- ❌ Don't rely on Quarkus Flow's taskExecutionId for identity (it changes per event)
 - ❌ Don't assume Quarkus Flow will fix this (working as designed for their use case)
-- ❌ Don't remove `task_execution_id` column (GraphQL API still needs it)
 
 ### Metrics & Observability
 
@@ -474,16 +474,19 @@ data-index.metrics.transform.poll-interval=30s
 
 **Database columns → JPA Entity → GraphQL:**
 ```
-task_instances.task_execution_id → TaskInstanceEntity.taskExecutionId → TaskExecution.id
-task_instances.start             → TaskInstanceEntity.start          → TaskExecution.start (@JsonProperty("startDate"))
-task_instances.end               → TaskInstanceEntity.end            → TaskExecution.end (@JsonProperty("endDate"))
-task_instances.input             → TaskInstanceEntity.input          → TaskExecution.input (@Ignore, exposed via getInputData())
+task_instances.instance_id   → TaskInstanceEntity.instanceId   → TaskExecution.instanceId
+task_instances.task_position → TaskInstanceEntity.taskPosition → TaskExecution.taskPosition
+(derived ID)                 → TaskInstanceEntity.getId()      → TaskExecution.getId() (returns "instanceId:taskPosition")
+task_instances.start         → TaskInstanceEntity.start        → TaskExecution.start (@JsonProperty("startDate"))
+task_instances.end           → TaskInstanceEntity.end          → TaskExecution.end (@JsonProperty("endDate"))
+task_instances.input         → TaskInstanceEntity.input        → TaskExecution.input (@Ignore, exposed via getInputData())
 ```
 
 **IMPORTANT:**
+- TaskExecution.id is derived from instanceId + taskPosition (no database column)
 - Database uses `start`/`end` (reserved SQL keywords, quoted)
 - GraphQL uses `startDate`/`endDate` (via `@JsonProperty`)
-- Never use old names: `enter`/`exit`, `triggerTime`/`leaveTime`
+- Never use old names: `enter`/`exit`, `triggerTime`/`leaveTime`, `taskExecutionId`
 
 ### Error Handling
 

data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/TaskExecutionProcessor.java

@@ -55,11 +55,6 @@ public void processBatch(List<TaskExecution> events) {
         log.debug("Processing task batch size: {}", events.size());
 
         try {
-            for (TaskExecution event : events) {
-                Objects.requireNonNull(event, "event cannot be null");
-                event.setId(generateTaskExecutionId(event));
-            }
-
             this.taskPersistence.persistBatch(events);
 
             log.debug("Successfully processed task batch size: {}", events.size());
@@ -68,10 +63,4 @@ public void processBatch(List<TaskExecution> events) {
             throw new ProcessEventFailedException("Failed to process task event batch", e);
         }
     }
-
-    private String generateTaskExecutionId(TaskExecution taskExecutionEvent) {
-        return taskExecutionEvent.getInstanceId()
-                + ":"
-                + taskExecutionEvent.getTaskPosition();
-    }
 }

data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/processor/persistence/TaskPersistence.java

@@ -140,33 +140,31 @@ private void tryCreatePlaceholder(TaskExecution event, Connection conn) throws S
 
     private void setTaskParameters(PreparedStatement stmt, TaskExecution event) throws SQLException {
 
-        stmt.setString(1, event.getId());
-        stmt.setString(2, event.getInstanceId());
-        stmt.setString(3, event.getTaskName());
-        stmt.setString(4, event.getTaskPosition());
-        stmt.setString(5, event.getStatus());
-        stmt.setObject(6, Optional.ofNullable(event.getStart()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
-        stmt.setObject(7, Optional.ofNullable(event.getEnd()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
-        stmt.setString(8, toJsonString(event.getInput()));
-        stmt.setString(9, toJsonString(event.getOutput()));
-
+        stmt.setString(1, event.getInstanceId());
+        stmt.setString(2, event.getTaskName());
+        stmt.setString(3, event.getTaskPosition());
+        stmt.setString(4, event.getStatus());
+        stmt.setObject(5, Optional.ofNullable(event.getStart()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
+        stmt.setObject(6, Optional.ofNullable(event.getEnd()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
+        stmt.setString(7, toJsonString(event.getInput()));
+        stmt.setString(8, toJsonString(event.getOutput()));
 
         // Error fields
         if (event.getError() != null) {
-            stmt.setString(10, event.getError().getType());
-            stmt.setString(11, event.getError().getTitle());
-            stmt.setString(12, event.getError().getDetail());
-            stmt.setObject(13, event.getError().getStatus());
-            stmt.setString(14, event.getError().getInstance());
+            stmt.setString(9, event.getError().getType());
+            stmt.setString(10, event.getError().getTitle());
+            stmt.setString(11, event.getError().getDetail());
+            stmt.setObject(12, event.getError().getStatus());
+            stmt.setString(13, event.getError().getInstance());
         } else {
+            stmt.setNull(9, java.sql.Types.VARCHAR);
             stmt.setNull(10, java.sql.Types.VARCHAR);
             stmt.setNull(11, java.sql.Types.VARCHAR);
-            stmt.setNull(12, java.sql.Types.VARCHAR);
-            stmt.setNull(13, java.sql.Types.INTEGER);
-            stmt.setNull(14, java.sql.Types.VARCHAR);
+            stmt.setNull(12, java.sql.Types.INTEGER);
+            stmt.setNull(13, java.sql.Types.VARCHAR);
         }
 
-        stmt.setObject(15, Optional.ofNullable(event.getEventTimestamp()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
+        stmt.setObject(14, Optional.ofNullable(event.getEventTimestamp()).map(ZonedDateTime::toOffsetDateTime).orElse(null));
     }
 
     private String toJsonString(JsonNode node) {

data-index/data-index-ingestion/data-index-ingestion-kafka-processor/src/main/resources/sql/task-instance-upsert.sql

@@ -1,9 +1,9 @@
 INSERT INTO task_instances (
-  task_execution_id, instance_id, task_name, task_position, status,
+  instance_id, task_name, task_position, status,
   start, "end", input, output,
   error_type, error_title, error_detail, error_status, error_instance,
   last_event_time, created_at, updated_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?, ?, ?, ?, ?, ?, NOW(), NOW())
+) VALUES (?, ?, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?, ?, ?, ?, ?, ?, NOW(), NOW())
 ON CONFLICT (instance_id, task_position) DO UPDATE SET
   instance_id = COALESCE(EXCLUDED.instance_id, task_instances.instance_id),
   task_name = COALESCE(EXCLUDED.task_name, task_instances.task_name),

data-index/data-index-ingestion/data-index-ingestion-kafka-service/src/main/java/org/kubesmarts/logic/dataindex/ingestion/kafka/service/Mapper.java

@@ -24,9 +24,6 @@
 import org.kubesmarts.logic.dataindex.model.TaskExecution;
 import org.kubesmarts.logic.dataindex.model.WorkflowInstance;
 import org.kubesmarts.logic.dataindex.model.WorkflowInstanceStatus;
-import java.time.OffsetDateTime;
-import java.util.UUID;
-
 public final class Mapper {
 
     private Mapper() {}
@@ -91,7 +88,6 @@ public static TaskExecution mapTaskExecutionEvent(CloudEvent cloudEvent, TaskCED
         taskExecution.setInstanceId(instanceId);
         taskExecution.setEventTimestamp(cloudEvent.getTime().toZonedDateTime());
         taskExecution.setStatus(status);
-        taskExecution.setId(generateTaskExecutionId(instanceId, taskPosition, cloudEvent.getTime()));
         taskExecution.setTaskPosition(taskPosition);
         taskExecution.setTaskName(taskPosition.substring(taskPosition.lastIndexOf("/") + 1));
 
@@ -141,9 +137,4 @@ private static Error mapError(WorkflowError workflowError) {
         error.setInstance(workflowError.instance());
         return error;
     }
-
-    private static String generateTaskExecutionId(String instanceId, String taskPosition, OffsetDateTime time) {
-        String composite = instanceId + ":" + taskPosition + ":" + time.toInstant().toEpochMilli();
-        return UUID.nameUUIDFromBytes(composite.getBytes()).toString();
-    }
 }

data-index/data-index-integration-tests/data-index-integration-tests-postgresql/src/test/java/org/kubesmarts/logic/dataindex/graphql/WorkflowInstanceGraphQLApiTest.java

@@ -94,7 +94,6 @@ public void setupTestData() throws Exception {
         List<TaskInstanceEntity> tasks1 = new ArrayList<>();
 
         TaskInstanceEntity task1 = new TaskInstanceEntity();
-        task1.setTaskExecutionId("task-1-1");
         task1.setInstanceId(TEST_WORKFLOW_ID_1);
         task1.setTaskName("validateInput");
         task1.setTaskPosition("/do/0");
@@ -107,7 +106,6 @@ public void setupTestData() throws Exception {
         tasks1.add(task1);
 
         TaskInstanceEntity task2 = new TaskInstanceEntity();
-        task2.setTaskExecutionId("task-1-2");
         task2.setInstanceId(TEST_WORKFLOW_ID_1);
         task2.setTaskName("processData");
         task2.setTaskPosition("/do/1");
@@ -144,7 +142,6 @@ public void setupTestData() throws Exception {
         List<TaskInstanceEntity> tasks2 = new ArrayList<>();
 
         TaskInstanceEntity task3 = new TaskInstanceEntity();
-        task3.setTaskExecutionId("task-2-1");
         task3.setInstanceId(TEST_WORKFLOW_ID_2);
         task3.setTaskName("failingTask");
         task3.setTaskPosition("/do/0");

data-index/data-index-model/src/main/java/org/kubesmarts/logic/dataindex/model/TaskExecution.java

@@ -98,10 +98,25 @@ public class TaskExecution {
      */
     private ZonedDateTime eventTimestamp;
 
+    /**
+     * Get derived task execution ID.
+     * <p>ID is derived from instanceId + taskPosition (format: "instanceId:taskPosition")
+     * <p>No database column - computed from composite key
+     */
     public String getId() {
-        return id;
+        if (id != null) {
+            return id;
+        }
+        if (instanceId != null && taskPosition != null) {
+            return instanceId + ":" + taskPosition;
+        }
+        return null;
     }
 
+    /**
+     * Set task execution ID.
+     * <p>For deserialization only - ID is normally derived from instanceId + taskPosition
+     */
     public void setId(String id) {
         this.id = id;
     }

data-index/data-index-storage/data-index-storage-elasticsearch-schema/src/main/resources/elasticsearch/index-templates/task-events.json

@@ -31,14 +31,6 @@
             }
           }
         },
-        "taskExecutionId": {
-          "type": "text",
-          "fields": {
-            "keyword": {
-              "type": "keyword"
-            }
-          }
-        },
         "taskPosition": {
           "type": "text",
           "fields": {

data-index/data-index-storage/data-index-storage-migrations/src/main/resources/db/migration/V1__initial_schema.sql

@@ -64,7 +64,6 @@ CREATE INDEX IF NOT EXISTS idx_workflow_instances_start ON workflow_instances (s
 CREATE INDEX IF NOT EXISTS idx_workflow_instances_last_event_time ON workflow_instances (last_event_time DESC);
 
 CREATE TABLE IF NOT EXISTS task_instances (
-  task_execution_id VARCHAR(255),
   instance_id VARCHAR(255) NOT NULL,
   task_name VARCHAR(255),
   task_position VARCHAR(255) NOT NULL,
@@ -82,16 +81,14 @@ CREATE TABLE IF NOT EXISTS task_instances (
   created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
   updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
   -- Composite primary key: (instance_id, task_position) uniquely identifies a task
-  -- Quarkus Flow generates different taskExecutionId per event (started, completed, etc.)
-  -- Same workflow + same position = same task execution
+  -- TaskExecution.id is derived from instanceId + taskPosition (no column needed)
   PRIMARY KEY (instance_id, task_position),
   CONSTRAINT fk_task_instance_workflow FOREIGN KEY (instance_id) REFERENCES workflow_instances(id) ON DELETE CASCADE
 );
 
 CREATE INDEX IF NOT EXISTS idx_task_instances_instance_id ON task_instances (instance_id);
 CREATE INDEX IF NOT EXISTS idx_task_instances_status ON task_instances (status);
 CREATE INDEX IF NOT EXISTS idx_task_instances_last_event_time ON task_instances (last_event_time DESC);
-CREATE INDEX IF NOT EXISTS idx_task_instances_task_execution_id ON task_instances (task_execution_id);
 
 -- ============================================================================
 -- TRIGGER FUNCTIONS (Extract from JSONB and normalize with idempotency)
@@ -218,7 +215,6 @@ BEGIN
   -- instance already exists because workflow events have already created it.
   BEGIN
     INSERT INTO task_instances (
-      task_execution_id,
       instance_id,
       task_name,
       task_position,
@@ -236,7 +232,6 @@ BEGIN
       created_at,
       updated_at
     ) VALUES (
-      NEW.data->>'taskExecutionId',
       NEW.data->>'instanceId',
       NEW.data->>'taskName',
       NEW.data->>'taskPosition',
@@ -267,9 +262,6 @@ BEGIN
       NEW.time
     )
     ON CONFLICT (instance_id, task_position) DO UPDATE SET
-      -- Update task_execution_id to the latest (even though it changes per event)
-      task_execution_id = EXCLUDED.task_execution_id,
-
       -- Keep first non-null values (immutable fields)
       task_name = COALESCE(task_instances.task_name, EXCLUDED.task_name),
 
@@ -318,7 +310,6 @@ BEGIN
     ON CONFLICT (id) DO NOTHING;
 
     INSERT INTO task_instances (
-      task_execution_id,
       instance_id,
       task_name,
       task_position,
@@ -336,7 +327,6 @@ BEGIN
       created_at,
       updated_at
     ) VALUES (
-      NEW.data->>'taskExecutionId',
       NEW.data->>'instanceId',
       NEW.data->>'taskName',
       NEW.data->>'taskPosition',
@@ -367,9 +357,6 @@ BEGIN
       NEW.time
     )
     ON CONFLICT (instance_id, task_position) DO UPDATE SET
-      -- Update task_execution_id to the latest (even though it changes per event)
-      task_execution_id = EXCLUDED.task_execution_id,
-
       -- Keep first non-null values (immutable fields)
       task_name = COALESCE(task_instances.task_name, EXCLUDED.task_name),
 

data-index/data-index-storage/data-index-storage-postgresql/src/main/java/org/kubesmarts/logic/dataindex/storage/jpa/TaskExecutionJPAStorage.java

@@ -62,6 +62,44 @@ protected TaskExecutionJPAStorage() {
         super();
     }
 
+    /**
+     * Get task execution by derived ID.
+     * <p>ID format: "instanceId:taskPosition"
+     * <p>Parses the composite ID and queries by both instanceId and taskPosition.
+     *
+     * @param id Derived ID in format "instanceId:taskPosition"
+     * @return TaskExecution or null if not found
+     */
+    @Override
+    public TaskExecution get(String id) {
+        if (id == null || id.isEmpty()) {
+            return null;
+        }
+
+        // Parse composite ID: "instanceId:taskPosition"
+        int separatorIndex = id.indexOf(':');
+        if (separatorIndex == -1 || separatorIndex == 0 || separatorIndex == id.length() - 1) {
+            throw new IllegalArgumentException("Invalid task execution ID format. Expected 'instanceId:taskPosition', got: " + id);
+        }
+
+        String instanceId = id.substring(0, separatorIndex);
+        String taskPosition = id.substring(separatorIndex + 1);
+
+        // Query by composite key
+        List<TaskInstanceEntity> entities = em
+                .createQuery("SELECT t FROM TaskInstanceEntity t WHERE t.instanceId = :instanceId AND t.taskPosition = :taskPosition", TaskInstanceEntity.class)
+                .setParameter("instanceId", instanceId)
+                .setParameter("taskPosition", taskPosition)
+                .setMaxResults(1)
+                .getResultList();
+
+        if (entities.isEmpty()) {
+            return null;
+        }
+
+        return mapToModel.apply(entities.get(0));
+    }
+
     @Override
     public List<TaskExecution> findByWorkflowInstanceId(String workflowInstanceId) {
         List<TaskInstanceEntity> entities = em