Remove raw tables, Vector insert directly to normalized tables instead

[mathecruz]

Changes

Builds on #73. Vector now inserts directly into workflow_instances/task_instances instead of the raw staging tables, and the normalization triggers move onto those same tables (self-targeting INSERT ... ON CONFLICT DO UPDATE, guarded by pg_trigger_depth() to avoid recursion).

Closes #74

Load test

Load test script + results: mcruzdev/k6-scripts

MODE 1 k6 results: before vs after

metric	before	after
time_to_queryable avg	2343ms	2210ms
time_to_queryable p95	14051ms	12736ms
time_to_first_appearance avg	2335ms	2139ms
workflow_trigger_duration avg	25.8ms	15.4ms
never_appeared_failures	13	24
stuck_placeholder_failures	0	0

• Latency: after is on par or slightly better across the board.
• Open issue: never_appeared_failures roughly doubled (13→24 out of ~2200). Single trial each — needs repeat runs to confirm signal vs noise.

Follow-up: typed-column inserts (RATE=60, 90s stress)

metric	raw_event (self-targeting trigger)	typed columns
time_to_queryable avg	1.75s	849ms
time_to_queryable p95	9.67s	3.64s
never_appeared_failures	220	0

How-to test the changes

Below you can see how to test the entire change with a real Kubernetes cluster with KinD.

Step-by-step walkthrough to manually verify direct normalized-table inserts for MODE 1

Starting Colima:

colima stop

colima start --cpu 6 --memory 12 --disk 60

1. Create the KIND cluster

kind create cluster --name data-index-test --config data-index/helm/data-index/kind-cluster.yaml

kubectl cluster-info --context kind-data-index-test

Maps localhost:30080 (GraphQL), localhost:30082 (workflow app), localhost:30432 (Postgres).

2. Build + load the MODE 1 images

( cd data-index && MODE=mode1 bash scripts/e2e/common-setup.sh )

Check docker images:

docker exec data-index-test-control-plane crictl images | grep kubesmarts

3. Install the chart (Postgres + Vector + triggers)

helm upgrade --install data-index data-index/helm/data-index -n default --create-namespace -f data-index/helm/data-index/values-mode1.yaml

Confirm one vector DaemonSet, no fluentbit:

kubectl get ds -A

kubectl get cm -n logging vector-config -o jsonpath='{.data.vector\.yaml}' | head -3

4. Wait for pods

PostgreSQL:

kubectl wait -n postgresql --for=condition=ready pod/postgresql-0 --timeout=180s

Vector:

kubectl wait -n logging --for=condition=ready pod -l app=vector --timeout=180s

Data Index Service:

kubectl wait -n default --for=condition=ready pod -l app=data-index-service --timeout=300s

Workflow Test App:

kubectl wait -n workflows --for=condition=ready pod -l app=workflow-test-app --timeout=300s

kubectl get pods -A | grep -E 'postgresql|data-index|vector|workflow'

5. Verify the infrastructure

bash data-index/scripts/e2e/verify-infrastructure.sh mode1

Expect: Vector running, GraphQL ready, 2 tables (workflow_instances, task_instances — no more workflow_events_raw/task_events_raw), 2 triggers, now defined directly on those 2 tables instead of on the removed raw tables.

kubectl logs -n logging -l app=vector | grep -iE 'healthcheck|postgres|error'

NOTE: If you got some "Timed out." you can restart the daemonset/vector:

kubectl rollout restart daemonset/vector -n logging

6. Turn on event tracing (optional)

kubectl set env daemonset/vector -n logging DEBUG_EVENTS=true

kubectl rollout status ds/vector -n logging

kubectl logs -n logging -l app=vector -f

(leave that streaming in another terminal)

7. Trigger a workflow

curl -s -XPOST localhost:30082/test-workflows/hello-world -H 'content-type: application/json' -d '{"name":"neo"}'

Traced Vector output should now show lines shaped {"raw_event":{...}} (previously {"tag":"...","time":"...","data":{...}}) — Vector no longer computes a tag/time envelope, it hands the untouched event straight to the sink under raw_event.

8. Verify normalized tables (written directly by Vector, populated by the self-targeting triggers)

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "select id,name,status,started_at,ended_at,raw_event from workflow_instances;"

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "select instance_id,task,task_name,status,raw_event from task_instances;"

Expect the workflow row status = COMPLETED and its task rows, each raw_event holding the most recently applied event's JSON (latest event only, not full history — that's the trade-off for dropping the raw staging tables).

Confirm the raw tables are really gone:

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "\dt" | grep -c _raw

Expect 0.

9. Verify the GraphQL API

curl -s localhost:30080/graphql -H 'content-type: application/json' -d '{"query":"{ getWorkflowInstances(limit:5){ id name status startedAt endedAt taskExecutions{ task taskName status startedAt endedAt } } }"}' | python3 -m json.tool

WorkflowInstance / TaskExecution fields: id name namespace version status startedAt endedAt lastUpdate eventTimestamp inputData outputData error{ ... } taskExecutions{ ... }. Introspect with { __type(name:"WorkflowInstance"){ fields{ name } } }. No GraphQL schema changes in this PR — raw_event isn't exposed.

10. Idempotency spot-check (optional)

for i in 1 2 3; do curl -s -XPOST localhost:30082/test-workflows/hello-world -H 'content-type: application/json' -d '{"name":"neo"}' >/dev/null; done; sleep 5

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "select instance_id, task, count(*) from task_instances group by 1,2 having count(*) > 1;"

Expect: 0 rows — the trigger's INSERT ... ON CONFLICT DO UPDATE merges repeated events into the same row instead of inserting duplicates.

11. Out-of-order spot-check (optional)

Confirms a task event that beats its workflow's own event still resolves correctly (via the trigger's proactive placeholder-workflow insert):

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "
  insert into task_instances (raw_event) values ('{\"instanceId\":\"manual-test-1\",\"taskPosition\":\"/do/0\",\"taskName\":\"step1\",\"status\":\"RUNNING\",\"startTime\":1700000000,\"timestamp\":1700000000}'::jsonb);
  select id, status from workflow_instances where id='manual-test-1';
  select instance_id, task, task_name, status from task_instances where instance_id='manual-test-1';
"

12. Concurrency spot-check (optional but recommended)

This is the actual scenario that broke in earlier testing of this PR: the postgres_workflow and postgres_task sinks are separate, concurrent connections, and a task event racing a batch of workflow events for the same instanceId produced a duplicate key value violates unique constraint "workflow_instances_pkey" error that silently dropped the whole workflow-event batch (visible as an empty/placeholder-only workflow_instances row despite task_instances having real data). Confirms the fix (INSERT ... ON CONFLICT + pg_trigger_depth() guard) holds under real concurrent writes:

(kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "
  insert into task_instances (raw_event) values ('{\"instanceId\":\"concurrency-test-1\",\"taskPosition\":\"/do/0\",\"taskName\":\"step1\",\"status\":\"COMPLETED\",\"startTime\":1700000000,\"endTime\":1700000001,\"timestamp\":1700000001}'::jsonb);
") &
(kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "
  insert into workflow_instances (raw_event) values
  ('{\"instanceId\":\"concurrency-test-1\",\"status\":\"RUNNING\",\"startTime\":1700000000,\"timestamp\":1700000000,\"workflowName\":\"race-test\"}'::jsonb),
  ('{\"instanceId\":\"concurrency-test-1\",\"status\":\"COMPLETED\",\"endTime\":1700000001,\"timestamp\":1700000001,\"output\":{\"r\":1}}'::jsonb);
") &
wait

kubectl exec -n postgresql postgresql-0 -- psql -U dataindex -d dataindex -c "select id,name,status,started_at,ended_at,output from workflow_instances where id='concurrency-test-1';"

Expect no duplicate key error from either background job, and a single fully-populated workflow_instances row (status = COMPLETED, name = race-test, output = {"r": 1}) — not an empty placeholder.
Expect a placeholder workflow_instances row (namespace/status NULL) and a normal task_instances row.

13. Cleanup

kind delete cluster --name data-index-test

Config source of truth: data-index/collectors/vector/mode1-postgresql/vector.yaml.
Full reference: data-index/data-index-docs/modules/ROOT/pages/deployment/vector-config.adoc.

[mathecruz]

Working to solve the open issue mentioned in the PR's description.

[ricardozanini]

Why can't vector insert into the fields directly instead?

[mathecruz]

The vector postgresql sink does not support ON CONFLIC the UPSERT approach (only INSERT for now).

[mathecruz]

Working to solve the open issue mentioned in the PR's description.

Investigated the never_appeared_failures, no data loss. It was not possible to reproduce at the tested load (RATE=20) in two reruns. I think it is safe to merge.

[ricardozanini]

We can reuse the same since we haven't release it yet.

[ricardozanini]

The idea is not to add to a raw column, but to map directly to the table here. If it's to keep the raw data, we keep the table. Keeping the table is complex, since we will need to implement a GC, or the tables will grow indefinitely with duplicated data.