diff --git a/cmd/hatchet-migrate/migrate/migrations/20261013191212_v1_0_172.sql b/cmd/hatchet-migrate/migrate/migrations/20261013191212_v1_0_172.sql new file mode 100644 index 0000000000..cb294476b5 --- /dev/null +++ b/cmd/hatchet-migrate/migrate/migrations/20261013191212_v1_0_172.sql @@ -0,0 +1,9 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TABLE v1_task_events_olap ADD COLUMN is_dag_orchestrator BOOLEAN NOT NULL DEFAULT FALSE; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE v1_task_events_olap DROP COLUMN is_dag_orchestrator; +-- +goose StatementEnd diff --git a/internal/services/controllers/olap/controller.go b/internal/services/controllers/olap/controller.go index d72cdaedbb..cd9eab6133 100644 --- a/internal/services/controllers/olap/controller.go +++ b/internal/services/controllers/olap/controller.go @@ -1108,6 +1108,7 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t WorkerID: workerId, AdditionalEventMessage: sqlchelpers.TextFromStr(eventMessages[i]), ExternalID: eventExternalIds[i], + IsDagOrchestrator: taskIdsToMetas[taskId].IsDagOrchestrator, } switch eventTypes[i] { diff --git a/pkg/repository/sqlcv1/copyfrom.go b/pkg/repository/sqlcv1/copyfrom.go index 6544b9d293..c9d9eeb5b3 100644 --- a/pkg/repository/sqlcv1/copyfrom.go +++ b/pkg/repository/sqlcv1/copyfrom.go @@ -156,6 +156,7 @@ func (r iteratorForCreateTaskEventsOLAP) Values() ([]interface{}, error) { r.rows[0].AdditionalEventMessage, r.rows[0].ExternalID, r.rows[0].DurableInvocationCount, + r.rows[0].IsDagOrchestrator, }, nil } @@ -164,7 +165,7 @@ func (r iteratorForCreateTaskEventsOLAP) Err() error { } func (q *Queries) CreateTaskEventsOLAP(ctx context.Context, db DBTX, arg []CreateTaskEventsOLAPParams) (int64, error) { - return db.CopyFrom(ctx, []string{"v1_task_events_olap"}, []string{"tenant_id", "task_id", "task_inserted_at", "event_type", "workflow_id", "event_timestamp", "readable_status", "retry_count", "error_message", "output", "worker_id", "additional__event_data", "additional__event_message", "external_id", "durable_invocation_count"}, &iteratorForCreateTaskEventsOLAP{rows: arg}) + return db.CopyFrom(ctx, []string{"v1_task_events_olap"}, []string{"tenant_id", "task_id", "task_inserted_at", "event_type", "workflow_id", "event_timestamp", "readable_status", "retry_count", "error_message", "output", "worker_id", "additional__event_data", "additional__event_message", "external_id", "durable_invocation_count", "is_dag_orchestrator"}, &iteratorForCreateTaskEventsOLAP{rows: arg}) } // iteratorForInsertLogLine implements pgx.CopyFromSource. diff --git a/pkg/repository/sqlcv1/models.go b/pkg/repository/sqlcv1/models.go index d716000aaa..724f60155c 100644 --- a/pkg/repository/sqlcv1/models.go +++ b/pkg/repository/sqlcv1/models.go @@ -3971,6 +3971,7 @@ type V1TaskEventsOlap struct { AdditionalEventData pgtype.Text `json:"additional__event_data"` AdditionalEventMessage pgtype.Text `json:"additional__event_message"` DurableInvocationCount int32 `json:"durable_invocation_count"` + IsDagOrchestrator bool `json:"is_dag_orchestrator"` } type V1TaskEventsOlapTmp struct { diff --git a/pkg/repository/sqlcv1/olap.sql b/pkg/repository/sqlcv1/olap.sql index d1d53c2639..b1588406e4 100644 --- a/pkg/repository/sqlcv1/olap.sql +++ b/pkg/repository/sqlcv1/olap.sql @@ -241,7 +241,8 @@ INSERT INTO v1_task_events_olap ( additional__event_data, additional__event_message, external_id, - durable_invocation_count + durable_invocation_count, + is_dag_orchestrator ) VALUES ( $1, $2, @@ -257,7 +258,8 @@ INSERT INTO v1_task_events_olap ( $12, $13, $14, - $15 + $15, + $16 ); -- name: ReadTaskByExternalID :one diff --git a/pkg/repository/sqlcv1/olap.sql.go b/pkg/repository/sqlcv1/olap.sql.go index 44f0085052..86d575206b 100644 --- a/pkg/repository/sqlcv1/olap.sql.go +++ b/pkg/repository/sqlcv1/olap.sql.go @@ -677,6 +677,7 @@ type CreateTaskEventsOLAPParams struct { AdditionalEventMessage pgtype.Text `json:"additional__event_message"` ExternalID uuid.UUID `json:"external_id"` DurableInvocationCount int32 `json:"durable_invocation_count"` + IsDagOrchestrator bool `json:"is_dag_orchestrator"` } const createV1PayloadOLAPCutoverTemporaryTable = `-- name: CreateV1PayloadOLAPCutoverTemporaryTable :exec @@ -2463,7 +2464,7 @@ WITH input AS ( JOIN v1_dags_olap d ON (r.id, r.inserted_at) = (d.id, d.inserted_at) WHERE r.tenant_id = $4::uuid AND r.kind = 'DAG' ), relevant_events AS ( - SELECT r.run_id, e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count + SELECT r.run_id, e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator FROM runs r JOIN v1_dag_to_task_olap dt ON (r.dag_id, r.inserted_at) = (dt.dag_id, dt.dag_inserted_at) JOIN v1_task_events_olap e ON (e.task_id, e.task_inserted_at) = (dt.task_id, dt.task_inserted_at) @@ -2688,7 +2689,7 @@ WITH selected_retry_count AS ( LIMIT 1 ), relevant_events AS ( SELECT - tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count + tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count, is_dag_orchestrator FROM v1_task_events_olap WHERE @@ -2932,7 +2933,7 @@ WITH input AS ( t.tenant_id = $4::uuid ), relevant_events AS ( SELECT - e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count + e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator FROM v1_task_events_olap e JOIN @@ -3464,7 +3465,7 @@ WITH runs AS ( AND lt.task_id IS NOT NULL ), relevant_events AS ( SELECT - e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count + e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator FROM runs r JOIN v1_dag_to_task_olap dt ON r.dag_id = dt.dag_id AND r.inserted_at = dt.dag_inserted_at JOIN v1_task_events_olap e ON (e.task_id, e.task_inserted_at) = (dt.task_id, dt.task_inserted_at) @@ -3476,7 +3477,7 @@ WITH runs AS ( UNION ALL SELECT - e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count + e.tenant_id, e.id, e.inserted_at, e.external_id, e.task_id, e.task_inserted_at, e.event_type, e.workflow_id, e.event_timestamp, e.readable_status, e.retry_count, e.error_message, e.output, e.worker_id, e.additional__event_data, e.additional__event_message, e.durable_invocation_count, e.is_dag_orchestrator FROM runs r JOIN v1_task_events_olap e ON e.task_id = r.task_id AND e.task_inserted_at = r.inserted_at WHERE @@ -3655,7 +3656,7 @@ WITH inputs AS ( ORDER BY inserted_at, id FOR UPDATE ), relevant_events AS MATERIALIZED ( - SELECT tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count + SELECT tenant_id, id, inserted_at, external_id, task_id, task_inserted_at, event_type, workflow_id, event_timestamp, readable_status, retry_count, error_message, output, worker_id, additional__event_data, additional__event_message, durable_invocation_count, is_dag_orchestrator FROM v1_task_events_olap WHERE (task_id, task_inserted_at) IN (SELECT task_id, task_inserted_at FROM inputs) diff --git a/sql/schema/v1-olap.sql b/sql/schema/v1-olap.sql index 8bd427af04..b520dab33d 100644 --- a/sql/schema/v1-olap.sql +++ b/sql/schema/v1-olap.sql @@ -406,6 +406,7 @@ CREATE TABLE v1_task_events_olap ( additional__event_data TEXT, additional__event_message TEXT, durable_invocation_count INT NOT NULL DEFAULT 0, + is_dag_orchestrator BOOLEAN NOT NULL DEFAULT FALSE, PRIMARY KEY (task_id, task_inserted_at, id) ) PARTITION BY RANGE (task_inserted_at);