mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2026-03-01 05:19:57 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: revert previous migration for new build of 0.52.0 (#1072)

Browse Source 
* fix: revert previous migration for new build of 0.52.0

* also remove identityId
This commit is contained in:
abelanger5
2024-11-25 14:03:36 -05:00
committed by GitHub
parent 20c49c9638
commit fbbe02fa33
11 changed files with 100 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
pkg/repository/prisma/dbsqlc/events.sql.go
View File

@@ -159,7 +159,7 @@ INSERT INTO "Event" (
$7::uuid,
$8::jsonb,
$9::jsonb
) RETURNING id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder", "identityId"
) RETURNING id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder"
`
type CreateEventParams struct {
@@ -198,7 +198,6 @@ func (q *Queries) CreateEvent(ctx context.Context, db DBTX, arg CreateEventParam
&i.Data,
&i.AdditionalMetadata,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
@@ -236,7 +235,7 @@ type CreateEventsParams struct {
const getEventForEngine = `-- name: GetEventForEngine :one
SELECT
id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder", "identityId"
id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder"
FROM
"Event"
WHERE
@@ -258,7 +257,6 @@ func (q *Queries) GetEventForEngine(ctx context.Context, db DBTX, id pgtype.UUID
&i.Data,
&i.AdditionalMetadata,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
@@ -304,7 +302,7 @@ func (q *Queries) GetEventsForRange(ctx context.Context, db DBTX) ([]*GetEventsF
}
const getInsertedEvents = `-- name: GetInsertedEvents :many
SELECT id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder", "identityId" FROM "Event"
SELECT id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder" FROM "Event"
WHERE "id" = ANY($1::uuid[])
ORDER BY "insertOrder" ASC
`
@@ -329,7 +327,6 @@ func (q *Queries) GetInsertedEvents(ctx context.Context, db DBTX, ids []pgtype.U
&i.Data,
&i.AdditionalMetadata,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -442,7 +439,7 @@ event_run_counts AS (
events."id"
)
SELECT
events.id, events."createdAt", events."updatedAt", events."deletedAt", events.key, events."tenantId", events."replayedFromId", events.data, events."additionalMetadata", events."insertOrder", events."identityId",
events.id, events."createdAt", events."updatedAt", events."deletedAt", events.key, events."tenantId", events."replayedFromId", events.data, events."additionalMetadata", events."insertOrder",
COALESCE(erc.pendingRuns, 0) AS pendingRuns,
COALESCE(erc.queuedRuns, 0) AS queuedRuns,
COALESCE(erc.runningRuns, 0) AS runningRuns,
@@ -512,7 +509,6 @@ func (q *Queries) ListEvents(ctx context.Context, db DBTX, arg ListEventsParams)
&i.Event.Data,
&i.Event.AdditionalMetadata,
&i.Event.InsertOrder,
&i.Event.IdentityId,
&i.Pendingruns,
&i.Queuedruns,
&i.Runningruns,
@@ -531,7 +527,7 @@ func (q *Queries) ListEvents(ctx context.Context, db DBTX, arg ListEventsParams)
const listEventsByIDs = `-- name: ListEventsByIDs :many
SELECT
id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder", "identityId"
id, "createdAt", "updatedAt", "deletedAt", key, "tenantId", "replayedFromId", data, "additionalMetadata", "insertOrder"
FROM
"Event" as events
WHERE
@@ -565,7 +561,6 @@ func (q *Queries) ListEventsByIDs(ctx context.Context, db DBTX, arg ListEventsBy
&i.Data,
&i.AdditionalMetadata,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}

7
pkg/repository/prisma/dbsqlc/job_runs.sql.go
View File

@@ -183,7 +183,7 @@ WITH steps AS (
jr."workflowRunId" = $1::uuid
)
SELECT
jr.id, jr."createdAt", jr."updatedAt", jr."deletedAt", jr."tenantId", jr."jobId", jr."tickerId", jr.status, jr.result, jr."startedAt", jr."finishedAt", jr."timeoutAt", jr."cancelledAt", jr."cancelledReason", jr."cancelledError", jr."workflowRunId", jr."identityId",
jr.id, jr."createdAt", jr."updatedAt", jr."deletedAt", jr."tenantId", jr."jobId", jr."tickerId", jr.status, jr.result, jr."startedAt", jr."finishedAt", jr."timeoutAt", jr."cancelledAt", jr."cancelledReason", jr."cancelledError", jr."workflowRunId",
j.id, j."createdAt", j."updatedAt", j."deletedAt", j."tenantId", j."workflowVersionId", j.name, j.description, j.timeout, j.kind
FROM "JobRun" jr
JOIN "Job" j
@@ -214,7 +214,6 @@ type ListJobRunsForWorkflowRunFullRow struct {
CancelledReason pgtype.Text `json:"cancelledReason"`
CancelledError pgtype.Text `json:"cancelledError"`
WorkflowRunId pgtype.UUID `json:"workflowRunId"`
IdentityId pgtype.Int8 `json:"identityId"`
Job Job `json:"job"`
}
@@ -244,7 +243,6 @@ func (q *Queries) ListJobRunsForWorkflowRunFull(ctx context.Context, db DBTX, ar
&i.CancelledReason,
&i.CancelledError,
&i.WorkflowRunId,
&i.IdentityId,
&i.Job.ID,
&i.Job.CreatedAt,
&i.Job.UpdatedAt,
@@ -393,7 +391,7 @@ const updateJobRunStatus = `-- name: UpdateJobRunStatus :one
UPDATE "JobRun"
SET "status" = $1::"JobRunStatus"
WHERE "id" = $2::uuid AND "tenantId" = $3::uuid
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId", "identityId"
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId"
`
type UpdateJobRunStatusParams struct {
@@ -422,7 +420,6 @@ func (q *Queries) UpdateJobRunStatus(ctx context.Context, db DBTX, arg UpdateJob
&i.CancelledReason,
&i.CancelledError,
&i.WorkflowRunId,
&i.IdentityId,
)
return &i, err
}

88
pkg/repository/prisma/dbsqlc/models.go
View File

@@ -1165,7 +1165,6 @@ type Event struct {
Data []byte `json:"data"`
AdditionalMetadata []byte `json:"additionalMetadata"`
InsertOrder pgtype.Int4 `json:"insertOrder"`
IdentityId pgtype.Int8 `json:"identityId"`
}
type EventKey struct {
@@ -1237,18 +1236,16 @@ type JobRun struct {
CancelledReason pgtype.Text `json:"cancelledReason"`
CancelledError pgtype.Text `json:"cancelledError"`
WorkflowRunId pgtype.UUID `json:"workflowRunId"`
IdentityId pgtype.Int8 `json:"identityId"`
}
type JobRunLookupData struct {
ID pgtype.UUID `json:"id"`
CreatedAt pgtype.Timestamp `json:"createdAt"`
UpdatedAt pgtype.Timestamp `json:"updatedAt"`
DeletedAt pgtype.Timestamp `json:"deletedAt"`
JobRunId pgtype.UUID `json:"jobRunId"`
TenantId pgtype.UUID `json:"tenantId"`
Data []byte `json:"data"`
IdentityId pgtype.Int8 `json:"identityId"`
ID pgtype.UUID `json:"id"`
CreatedAt pgtype.Timestamp `json:"createdAt"`
UpdatedAt pgtype.Timestamp `json:"updatedAt"`
DeletedAt pgtype.Timestamp `json:"deletedAt"`
JobRunId pgtype.UUID `json:"jobRunId"`
TenantId pgtype.UUID `json:"tenantId"`
Data []byte `json:"data"`
}
type Lease struct {
@@ -1443,7 +1440,6 @@ type StepRun struct {
Queue string `json:"queue"`
Priority pgtype.Int4 `json:"priority"`
InternalRetryCount int32 `json:"internalRetryCount"`
IdentityId int64 `json:"identityId"`
}
type StepRunEvent struct {
@@ -1491,74 +1487,6 @@ type StepRunResultArchive struct {
RetryCount int32 `json:"retryCount"`
}
type StepRunStable struct {
ID pgtype.UUID `json:"id"`
CreatedAt pgtype.Timestamp `json:"createdAt"`
UpdatedAt pgtype.Timestamp `json:"updatedAt"`
DeletedAt pgtype.Timestamp `json:"deletedAt"`
TenantId pgtype.UUID `json:"tenantId"`
JobRunId pgtype.UUID `json:"jobRunId"`
StepId pgtype.UUID `json:"stepId"`
Order int64 `json:"order"`
WorkerId pgtype.UUID `json:"workerId"`
TickerId pgtype.UUID `json:"tickerId"`
Status StepRunStatus `json:"status"`
Input []byte `json:"input"`
Output []byte `json:"output"`
RequeueAfter pgtype.Timestamp `json:"requeueAfter"`
ScheduleTimeoutAt pgtype.Timestamp `json:"scheduleTimeoutAt"`
Error pgtype.Text `json:"error"`
StartedAt pgtype.Timestamp `json:"startedAt"`
FinishedAt pgtype.Timestamp `json:"finishedAt"`
TimeoutAt pgtype.Timestamp `json:"timeoutAt"`
CancelledAt pgtype.Timestamp `json:"cancelledAt"`
CancelledReason pgtype.Text `json:"cancelledReason"`
CancelledError pgtype.Text `json:"cancelledError"`
InputSchema []byte `json:"inputSchema"`
CallerFiles []byte `json:"callerFiles"`
GitRepoBranch pgtype.Text `json:"gitRepoBranch"`
RetryCount int32 `json:"retryCount"`
SemaphoreReleased bool `json:"semaphoreReleased"`
Queue string `json:"queue"`
Priority pgtype.Int4 `json:"priority"`
InternalRetryCount int32 `json:"internalRetryCount"`
IdentityId int64 `json:"identityId"`
}
type StepRunVolatile struct {
ID pgtype.UUID `json:"id"`
CreatedAt pgtype.Timestamp `json:"createdAt"`
UpdatedAt pgtype.Timestamp `json:"updatedAt"`
DeletedAt pgtype.Timestamp `json:"deletedAt"`
TenantId pgtype.UUID `json:"tenantId"`
JobRunId pgtype.UUID `json:"jobRunId"`
StepId pgtype.UUID `json:"stepId"`
Order int64 `json:"order"`
WorkerId pgtype.UUID `json:"workerId"`
TickerId pgtype.UUID `json:"tickerId"`
Status StepRunStatus `json:"status"`
Input []byte `json:"input"`
Output []byte `json:"output"`
RequeueAfter pgtype.Timestamp `json:"requeueAfter"`
ScheduleTimeoutAt pgtype.Timestamp `json:"scheduleTimeoutAt"`
Error pgtype.Text `json:"error"`
StartedAt pgtype.Timestamp `json:"startedAt"`
FinishedAt pgtype.Timestamp `json:"finishedAt"`
TimeoutAt pgtype.Timestamp `json:"timeoutAt"`
CancelledAt pgtype.Timestamp `json:"cancelledAt"`
CancelledReason pgtype.Text `json:"cancelledReason"`
CancelledError pgtype.Text `json:"cancelledError"`
InputSchema []byte `json:"inputSchema"`
CallerFiles []byte `json:"callerFiles"`
GitRepoBranch pgtype.Text `json:"gitRepoBranch"`
RetryCount int32 `json:"retryCount"`
SemaphoreReleased bool `json:"semaphoreReleased"`
Queue string `json:"queue"`
Priority pgtype.Int4 `json:"priority"`
InternalRetryCount int32 `json:"internalRetryCount"`
IdentityId int64 `json:"identityId"`
}
type StreamEvent struct {
ID int64 `json:"id"`
CreatedAt pgtype.Timestamp `json:"createdAt"`
@@ -1832,7 +1760,6 @@ type WorkflowRun struct {
Duration pgtype.Int8 `json:"duration"`
Priority pgtype.Int4 `json:"priority"`
InsertOrder pgtype.Int4 `json:"insertOrder"`
IdentityId pgtype.Int8 `json:"identityId"`
}
type WorkflowRunDedupe struct {
@@ -1868,7 +1795,6 @@ type WorkflowRunTriggeredBy struct {
Input []byte `json:"input"`
ParentId pgtype.UUID `json:"parentId"`
CronName pgtype.Text `json:"cronName"`
IdentityId pgtype.Int8 `json:"identityId"`
}
type WorkflowTag struct {

30
pkg/repository/prisma/dbsqlc/step_runs.sql.go
View File

@@ -758,7 +758,7 @@ func (q *Queries) GetFinalizedStepRuns(ctx context.Context, db DBTX, steprunids
const getLaterStepRuns = `-- name: GetLaterStepRuns :many
WITH RECURSIVE currStepRun AS (
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount", "identityId"
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount"
FROM "StepRun"
WHERE
"id" = $1::uuid
@@ -776,7 +776,7 @@ WITH RECURSIVE currStepRun AS (
JOIN childStepRuns csr ON sro."A" = csr."id"
)
SELECT
sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount", sr."identityId"
sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount"
FROM
"StepRun" sr
JOIN
@@ -823,7 +823,6 @@ func (q *Queries) GetLaterStepRuns(ctx context.Context, db DBTX, steprunid pgtyp
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -897,7 +896,7 @@ func (q *Queries) GetStepExpressions(ctx context.Context, db DBTX, stepid pgtype
const getStepRun = `-- name: GetStepRun :one
SELECT
"StepRun".id, "StepRun"."createdAt", "StepRun"."updatedAt", "StepRun"."deletedAt", "StepRun"."tenantId", "StepRun"."jobRunId", "StepRun"."stepId", "StepRun"."order", "StepRun"."workerId", "StepRun"."tickerId", "StepRun".status, "StepRun".input, "StepRun".output, "StepRun"."requeueAfter", "StepRun"."scheduleTimeoutAt", "StepRun".error, "StepRun"."startedAt", "StepRun"."finishedAt", "StepRun"."timeoutAt", "StepRun"."cancelledAt", "StepRun"."cancelledReason", "StepRun"."cancelledError", "StepRun"."inputSchema", "StepRun"."callerFiles", "StepRun"."gitRepoBranch", "StepRun"."retryCount", "StepRun"."semaphoreReleased", "StepRun".queue, "StepRun".priority, "StepRun"."internalRetryCount", "StepRun"."identityId"
"StepRun".id, "StepRun"."createdAt", "StepRun"."updatedAt", "StepRun"."deletedAt", "StepRun"."tenantId", "StepRun"."jobRunId", "StepRun"."stepId", "StepRun"."order", "StepRun"."workerId", "StepRun"."tickerId", "StepRun".status, "StepRun".input, "StepRun".output, "StepRun"."requeueAfter", "StepRun"."scheduleTimeoutAt", "StepRun".error, "StepRun"."startedAt", "StepRun"."finishedAt", "StepRun"."timeoutAt", "StepRun"."cancelledAt", "StepRun"."cancelledReason", "StepRun"."cancelledError", "StepRun"."inputSchema", "StepRun"."callerFiles", "StepRun"."gitRepoBranch", "StepRun"."retryCount", "StepRun"."semaphoreReleased", "StepRun".queue, "StepRun".priority, "StepRun"."internalRetryCount"
FROM
"StepRun"
WHERE
@@ -939,7 +938,6 @@ func (q *Queries) GetStepRun(ctx context.Context, db DBTX, id pgtype.UUID) (*Ste
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
)
return &i, err
}
@@ -1578,7 +1576,7 @@ func (q *Queries) ListInitialStepRuns(ctx context.Context, db DBTX, jobrunid pgt
const listNonFinalChildStepRuns = `-- name: ListNonFinalChildStepRuns :many
WITH RECURSIVE currStepRun AS (
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount", "identityId"
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount"
FROM "StepRun"
WHERE
"id" = $1::uuid
@@ -1597,7 +1595,7 @@ WITH RECURSIVE currStepRun AS (
JOIN childStepRuns csr ON sro."A" = csr."id"
)
SELECT
sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount", sr."identityId"
sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount"
FROM
"StepRun" sr
JOIN
@@ -1648,7 +1646,6 @@ func (q *Queries) ListNonFinalChildStepRuns(ctx context.Context, db DBTX, stepru
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -2458,7 +2455,7 @@ SET
"cancelledError" = NULL
WHERE
"id" = $1::uuid
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId", "identityId"
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId"
`
func (q *Queries) ReplayStepRunResetJobRun(ctx context.Context, db DBTX, jobrunid pgtype.UUID) (*JobRun, error) {
@@ -2481,14 +2478,13 @@ func (q *Queries) ReplayStepRunResetJobRun(ctx context.Context, db DBTX, jobruni
&i.CancelledReason,
&i.CancelledError,
&i.WorkflowRunId,
&i.IdentityId,
)
return &i, err
}
const replayStepRunResetStepRuns = `-- name: ReplayStepRunResetStepRuns :many
WITH RECURSIVE currStepRun AS (
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount", "identityId"
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobRunId", "stepId", "order", "workerId", "tickerId", status, input, output, "requeueAfter", "scheduleTimeoutAt", error, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "inputSchema", "callerFiles", "gitRepoBranch", "retryCount", "semaphoreReleased", queue, priority, "internalRetryCount"
FROM "StepRun"
WHERE
"id" = $1::uuid
@@ -2526,7 +2522,7 @@ FROM
WHERE
sr."id" = csr."id" OR
sr."id" = $1::uuid
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount", sr."identityId"
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount"
`
type ReplayStepRunResetStepRunsParams struct {
@@ -2574,7 +2570,6 @@ func (q *Queries) ReplayStepRunResetStepRuns(ctx context.Context, db DBTX, arg R
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -2599,7 +2594,7 @@ SET
"error" = NULL
WHERE
"id" = $1::uuid
RETURNING "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder", "identityId"
RETURNING "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder"
`
func (q *Queries) ReplayStepRunResetWorkflowRun(ctx context.Context, db DBTX, workflowrunid pgtype.UUID) (*WorkflowRun, error) {
@@ -2626,7 +2621,6 @@ func (q *Queries) ReplayStepRunResetWorkflowRun(ctx context.Context, db DBTX, wo
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
@@ -2695,7 +2689,7 @@ SET
WHERE
sr."id" = ANY($1::uuid[]) AND
sr."tenantId" = $2::uuid
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount", sr."identityId"
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount"
`
type ResetStepRunsByIdsParams struct {
@@ -2743,7 +2737,6 @@ func (q *Queries) ResetStepRunsByIds(ctx context.Context, db DBTX, arg ResetStep
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -2796,7 +2789,7 @@ FROM
childStepRuns csr
WHERE
sr."id" = csr."id"
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount", sr."identityId"
RETURNING sr.id, sr."createdAt", sr."updatedAt", sr."deletedAt", sr."tenantId", sr."jobRunId", sr."stepId", sr."order", sr."workerId", sr."tickerId", sr.status, sr.input, sr.output, sr."requeueAfter", sr."scheduleTimeoutAt", sr.error, sr."startedAt", sr."finishedAt", sr."timeoutAt", sr."cancelledAt", sr."cancelledReason", sr."cancelledError", sr."inputSchema", sr."callerFiles", sr."gitRepoBranch", sr."retryCount", sr."semaphoreReleased", sr.queue, sr.priority, sr."internalRetryCount"
`
type ResolveLaterStepRunsParams struct {
@@ -2844,7 +2837,6 @@ func (q *Queries) ResolveLaterStepRuns(ctx context.Context, db DBTX, arg Resolve
&i.Queue,
&i.Priority,
&i.InternalRetryCount,
&i.IdentityId,
); err != nil {
return nil, err
}

68
pkg/repository/prisma/dbsqlc/workflow_runs.sql.go
View File

@@ -402,7 +402,7 @@ INSERT INTO "JobRunLookupData" (
'triggered_by', $5::text,
'steps', '{}'::jsonb
)
) RETURNING id, "createdAt", "updatedAt", "deletedAt", "jobRunId", "tenantId", data, "identityId"
) RETURNING id, "createdAt", "updatedAt", "deletedAt", "jobRunId", "tenantId", data
`
type CreateJobRunLookupDataParams struct {
@@ -430,7 +430,6 @@ func (q *Queries) CreateJobRunLookupData(ctx context.Context, db DBTX, arg Creat
&i.JobRunId,
&i.TenantId,
&i.Data,
&i.IdentityId,
)
return &i, err
}
@@ -467,7 +466,7 @@ SELECT
'steps', '{}'::jsonb
)
FROM input_data
RETURNING id, "createdAt", "updatedAt", "deletedAt", "jobRunId", "tenantId", data, "identityId"
RETURNING id, "createdAt", "updatedAt", "deletedAt", "jobRunId", "tenantId", data
`
type CreateJobRunLookupDatasParams struct {
@@ -501,7 +500,6 @@ func (q *Queries) CreateJobRunLookupDatas(ctx context.Context, db DBTX, arg Crea
&i.JobRunId,
&i.TenantId,
&i.Data,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -845,7 +843,7 @@ INSERT INTO "WorkflowRun" (
$8::uuid,
$9::jsonb,
$10::int
) RETURNING "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder", "identityId"
) RETURNING "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder"
`
type CreateWorkflowRunParams struct {
@@ -896,7 +894,6 @@ func (q *Queries) CreateWorkflowRun(ctx context.Context, db DBTX, arg CreateWork
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
@@ -1029,7 +1026,7 @@ INSERT INTO "WorkflowRunTriggeredBy" (
$5::text,
$6::text,
$7::uuid
) RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "eventId", "cronParentId", "cronSchedule", "scheduledId", input, "parentId", "cronName", "identityId"
) RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "eventId", "cronParentId", "cronSchedule", "scheduledId", input, "parentId", "cronName"
`
type CreateWorkflowRunTriggeredByParams struct {
@@ -1066,7 +1063,6 @@ func (q *Queries) CreateWorkflowRunTriggeredBy(ctx context.Context, db DBTX, arg
&i.Input,
&i.ParentId,
&i.CronName,
&i.IdentityId,
)
return &i, err
}
@@ -1110,7 +1106,7 @@ func (q *Queries) DeleteScheduledWorkflow(ctx context.Context, db DBTX, schedule
const getChildWorkflowRun = `-- name: GetChildWorkflowRun :one
SELECT
"createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder", "identityId"
"createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder"
FROM
"WorkflowRun"
WHERE
@@ -1160,14 +1156,13 @@ func (q *Queries) GetChildWorkflowRun(ctx context.Context, db DBTX, arg GetChild
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
const getChildWorkflowRunsByIndex = `-- name: GetChildWorkflowRunsByIndex :many
SELECT
wr."createdAt", wr."updatedAt", wr."deletedAt", wr."tenantId", wr."workflowVersionId", wr.status, wr.error, wr."startedAt", wr."finishedAt", wr."concurrencyGroupId", wr."displayName", wr.id, wr."childIndex", wr."childKey", wr."parentId", wr."parentStepRunId", wr."additionalMetadata", wr.duration, wr.priority, wr."insertOrder", wr."identityId"
wr."createdAt", wr."updatedAt", wr."deletedAt", wr."tenantId", wr."workflowVersionId", wr.status, wr.error, wr."startedAt", wr."finishedAt", wr."concurrencyGroupId", wr."displayName", wr.id, wr."childIndex", wr."childKey", wr."parentId", wr."parentStepRunId", wr."additionalMetadata", wr.duration, wr.priority, wr."insertOrder"
FROM
"WorkflowRun" wr
WHERE
@@ -1216,7 +1211,6 @@ func (q *Queries) GetChildWorkflowRunsByIndex(ctx context.Context, db DBTX, arg
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -1230,7 +1224,7 @@ func (q *Queries) GetChildWorkflowRunsByIndex(ctx context.Context, db DBTX, arg
const getChildWorkflowRunsByKey = `-- name: GetChildWorkflowRunsByKey :many
SELECT
wr."createdAt", wr."updatedAt", wr."deletedAt", wr."tenantId", wr."workflowVersionId", wr.status, wr.error, wr."startedAt", wr."finishedAt", wr."concurrencyGroupId", wr."displayName", wr.id, wr."childIndex", wr."childKey", wr."parentId", wr."parentStepRunId", wr."additionalMetadata", wr.duration, wr.priority, wr."insertOrder", wr."identityId"
wr."createdAt", wr."updatedAt", wr."deletedAt", wr."tenantId", wr."workflowVersionId", wr.status, wr.error, wr."startedAt", wr."finishedAt", wr."concurrencyGroupId", wr."displayName", wr.id, wr."childIndex", wr."childKey", wr."parentId", wr."parentStepRunId", wr."additionalMetadata", wr.duration, wr.priority, wr."insertOrder"
FROM
"WorkflowRun" wr
WHERE
@@ -1279,7 +1273,6 @@ func (q *Queries) GetChildWorkflowRunsByKey(ctx context.Context, db DBTX, arg Ge
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -1605,8 +1598,8 @@ func (q *Queries) GetStepsForWorkflowVersion(ctx context.Context, db DBTX, workf
const getWorkflowRun = `-- name: GetWorkflowRun :many
SELECT
runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder", runs."identityId",
runtriggers.id, runtriggers."createdAt", runtriggers."updatedAt", runtriggers."deletedAt", runtriggers."tenantId", runtriggers."eventId", runtriggers."cronParentId", runtriggers."cronSchedule", runtriggers."scheduledId", runtriggers.input, runtriggers."parentId", runtriggers."cronName", runtriggers."identityId",
runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder",
runtriggers.id, runtriggers."createdAt", runtriggers."updatedAt", runtriggers."deletedAt", runtriggers."tenantId", runtriggers."eventId", runtriggers."cronParentId", runtriggers."cronSchedule", runtriggers."scheduledId", runtriggers.input, runtriggers."parentId", runtriggers."cronName",
workflowversion.id, workflowversion."createdAt", workflowversion."updatedAt", workflowversion."deletedAt", workflowversion.version, workflowversion."order", workflowversion."workflowId", workflowversion.checksum, workflowversion."scheduleTimeout", workflowversion."onFailureJobId", workflowversion.sticky, workflowversion.kind, workflowversion."defaultPriority",
workflow."name" as "workflowName",
-- waiting on https://github.com/sqlc-dev/sqlc/pull/2858 for nullable fields
@@ -1682,7 +1675,6 @@ func (q *Queries) GetWorkflowRun(ctx context.Context, db DBTX, arg GetWorkflowRu
&i.WorkflowRun.Duration,
&i.WorkflowRun.Priority,
&i.WorkflowRun.InsertOrder,
&i.WorkflowRun.IdentityId,
&i.WorkflowRunTriggeredBy.ID,
&i.WorkflowRunTriggeredBy.CreatedAt,
&i.WorkflowRunTriggeredBy.UpdatedAt,
@@ -1695,7 +1687,6 @@ func (q *Queries) GetWorkflowRun(ctx context.Context, db DBTX, arg GetWorkflowRu
&i.WorkflowRunTriggeredBy.Input,
&i.WorkflowRunTriggeredBy.ParentId,
&i.WorkflowRunTriggeredBy.CronName,
&i.WorkflowRunTriggeredBy.IdentityId,
&i.WorkflowVersion.ID,
&i.WorkflowVersion.CreatedAt,
&i.WorkflowVersion.UpdatedAt,
@@ -1756,10 +1747,10 @@ func (q *Queries) GetWorkflowRunAdditionalMeta(ctx context.Context, db DBTX, arg
const getWorkflowRunById = `-- name: GetWorkflowRunById :one
SELECT
r."createdAt", r."updatedAt", r."deletedAt", r."tenantId", r."workflowVersionId", r.status, r.error, r."startedAt", r."finishedAt", r."concurrencyGroupId", r."displayName", r.id, r."childIndex", r."childKey", r."parentId", r."parentStepRunId", r."additionalMetadata", r.duration, r.priority, r."insertOrder", r."identityId",
r."createdAt", r."updatedAt", r."deletedAt", r."tenantId", r."workflowVersionId", r.status, r.error, r."startedAt", r."finishedAt", r."concurrencyGroupId", r."displayName", r.id, r."childIndex", r."childKey", r."parentId", r."parentStepRunId", r."additionalMetadata", r.duration, r.priority, r."insertOrder",
wv.id, wv."createdAt", wv."updatedAt", wv."deletedAt", wv.version, wv."order", wv."workflowId", wv.checksum, wv."scheduleTimeout", wv."onFailureJobId", wv.sticky, wv.kind, wv."defaultPriority",
w.id, w."createdAt", w."updatedAt", w."deletedAt", w."tenantId", w.name, w.description, w."isPaused",
tb.id, tb."createdAt", tb."updatedAt", tb."deletedAt", tb."tenantId", tb."eventId", tb."cronParentId", tb."cronSchedule", tb."scheduledId", tb.input, tb."parentId", tb."cronName", tb."identityId"
tb.id, tb."createdAt", tb."updatedAt", tb."deletedAt", tb."tenantId", tb."eventId", tb."cronParentId", tb."cronSchedule", tb."scheduledId", tb.input, tb."parentId", tb."cronName"
FROM
"WorkflowRun" r
JOIN
@@ -1800,7 +1791,6 @@ type GetWorkflowRunByIdRow struct {
Duration pgtype.Int8 `json:"duration"`
Priority pgtype.Int4 `json:"priority"`
InsertOrder pgtype.Int4 `json:"insertOrder"`
IdentityId pgtype.Int8 `json:"identityId"`
WorkflowVersion WorkflowVersion `json:"workflow_version"`
Workflow Workflow `json:"workflow"`
WorkflowRunTriggeredBy WorkflowRunTriggeredBy `json:"workflow_run_triggered_by"`
@@ -1830,7 +1820,6 @@ func (q *Queries) GetWorkflowRunById(ctx context.Context, db DBTX, arg GetWorkfl
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
&i.WorkflowVersion.ID,
&i.WorkflowVersion.CreatedAt,
&i.WorkflowVersion.UpdatedAt,
@@ -1864,17 +1853,16 @@ func (q *Queries) GetWorkflowRunById(ctx context.Context, db DBTX, arg GetWorkfl
&i.WorkflowRunTriggeredBy.Input,
&i.WorkflowRunTriggeredBy.ParentId,
&i.WorkflowRunTriggeredBy.CronName,
&i.WorkflowRunTriggeredBy.IdentityId,
)
return &i, err
}
const getWorkflowRunByIds = `-- name: GetWorkflowRunByIds :many
SELECT
r."createdAt", r."updatedAt", r."deletedAt", r."tenantId", r."workflowVersionId", r.status, r.error, r."startedAt", r."finishedAt", r."concurrencyGroupId", r."displayName", r.id, r."childIndex", r."childKey", r."parentId", r."parentStepRunId", r."additionalMetadata", r.duration, r.priority, r."insertOrder", r."identityId",
r."createdAt", r."updatedAt", r."deletedAt", r."tenantId", r."workflowVersionId", r.status, r.error, r."startedAt", r."finishedAt", r."concurrencyGroupId", r."displayName", r.id, r."childIndex", r."childKey", r."parentId", r."parentStepRunId", r."additionalMetadata", r.duration, r.priority, r."insertOrder",
wv.id, wv."createdAt", wv."updatedAt", wv."deletedAt", wv.version, wv."order", wv."workflowId", wv.checksum, wv."scheduleTimeout", wv."onFailureJobId", wv.sticky, wv.kind, wv."defaultPriority",
w.id, w."createdAt", w."updatedAt", w."deletedAt", w."tenantId", w.name, w.description, w."isPaused",
tb.id, tb."createdAt", tb."updatedAt", tb."deletedAt", tb."tenantId", tb."eventId", tb."cronParentId", tb."cronSchedule", tb."scheduledId", tb.input, tb."parentId", tb."cronName", tb."identityId"
tb.id, tb."createdAt", tb."updatedAt", tb."deletedAt", tb."tenantId", tb."eventId", tb."cronParentId", tb."cronSchedule", tb."scheduledId", tb.input, tb."parentId", tb."cronName"
FROM
"WorkflowRun" r
JOIN
@@ -1915,7 +1903,6 @@ type GetWorkflowRunByIdsRow struct {
Duration pgtype.Int8 `json:"duration"`
Priority pgtype.Int4 `json:"priority"`
InsertOrder pgtype.Int4 `json:"insertOrder"`
IdentityId pgtype.Int8 `json:"identityId"`
WorkflowVersion WorkflowVersion `json:"workflow_version"`
Workflow Workflow `json:"workflow"`
WorkflowRunTriggeredBy WorkflowRunTriggeredBy `json:"workflow_run_triggered_by"`
@@ -1951,7 +1938,6 @@ func (q *Queries) GetWorkflowRunByIds(ctx context.Context, db DBTX, arg GetWorkf
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
&i.WorkflowVersion.ID,
&i.WorkflowVersion.CreatedAt,
&i.WorkflowVersion.UpdatedAt,
@@ -1985,7 +1971,6 @@ func (q *Queries) GetWorkflowRunByIds(ctx context.Context, db DBTX, arg GetWorkf
&i.WorkflowRunTriggeredBy.Input,
&i.WorkflowRunTriggeredBy.ParentId,
&i.WorkflowRunTriggeredBy.CronName,
&i.WorkflowRunTriggeredBy.IdentityId,
); err != nil {
return nil, err
}
@@ -2044,7 +2029,7 @@ func (q *Queries) GetWorkflowRunStickyStateForUpdate(ctx context.Context, db DBT
}
const getWorkflowRunTrigger = `-- name: GetWorkflowRunTrigger :one
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "eventId", "cronParentId", "cronSchedule", "scheduledId", input, "parentId", "cronName", "identityId"
SELECT id, "createdAt", "updatedAt", "deletedAt", "tenantId", "eventId", "cronParentId", "cronSchedule", "scheduledId", input, "parentId", "cronName"
FROM
"WorkflowRunTriggeredBy"
WHERE
@@ -2073,13 +2058,12 @@ func (q *Queries) GetWorkflowRunTrigger(ctx context.Context, db DBTX, arg GetWor
&i.Input,
&i.ParentId,
&i.CronName,
&i.IdentityId,
)
return &i, err
}
const getWorkflowRunsInsertedInThisTxn = `-- name: GetWorkflowRunsInsertedInThisTxn :many
SELECT "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder", "identityId" FROM "WorkflowRun"
SELECT "createdAt", "updatedAt", "deletedAt", "tenantId", "workflowVersionId", status, error, "startedAt", "finishedAt", "concurrencyGroupId", "displayName", id, "childIndex", "childKey", "parentId", "parentStepRunId", "additionalMetadata", duration, priority, "insertOrder" FROM "WorkflowRun"
WHERE xmin::text = (txid_current() % (2^32)::bigint)::text
AND ("createdAt" = CURRENT_TIMESTAMP::timestamp(3))
ORDER BY "insertOrder" ASC
@@ -2115,7 +2099,6 @@ func (q *Queries) GetWorkflowRunsInsertedInThisTxn(ctx context.Context, db DBTX)
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -2478,9 +2461,9 @@ func (q *Queries) ListWorkflowRunEventsByWorkflowRunId(ctx context.Context, db D
const listWorkflowRuns = `-- name: ListWorkflowRuns :many
SELECT
runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder", runs."identityId",
runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder",
workflow.id, workflow."createdAt", workflow."updatedAt", workflow."deletedAt", workflow."tenantId", workflow.name, workflow.description, workflow."isPaused",
runtriggers.id, runtriggers."createdAt", runtriggers."updatedAt", runtriggers."deletedAt", runtriggers."tenantId", runtriggers."eventId", runtriggers."cronParentId", runtriggers."cronSchedule", runtriggers."scheduledId", runtriggers.input, runtriggers."parentId", runtriggers."cronName", runtriggers."identityId",
runtriggers.id, runtriggers."createdAt", runtriggers."updatedAt", runtriggers."deletedAt", runtriggers."tenantId", runtriggers."eventId", runtriggers."cronParentId", runtriggers."cronSchedule", runtriggers."scheduledId", runtriggers.input, runtriggers."parentId", runtriggers."cronName",
workflowversion.id, workflowversion."createdAt", workflowversion."updatedAt", workflowversion."deletedAt", workflowversion.version, workflowversion."order", workflowversion."workflowId", workflowversion.checksum, workflowversion."scheduleTimeout", workflowversion."onFailureJobId", workflowversion.sticky, workflowversion.kind, workflowversion."defaultPriority",
-- waiting on https://github.com/sqlc-dev/sqlc/pull/2858 for nullable events field
events.id, events.key, events."createdAt", events."updatedAt"
@@ -2660,7 +2643,6 @@ func (q *Queries) ListWorkflowRuns(ctx context.Context, db DBTX, arg ListWorkflo
&i.WorkflowRun.Duration,
&i.WorkflowRun.Priority,
&i.WorkflowRun.InsertOrder,
&i.WorkflowRun.IdentityId,
&i.Workflow.ID,
&i.Workflow.CreatedAt,
&i.Workflow.UpdatedAt,
@@ -2681,7 +2663,6 @@ func (q *Queries) ListWorkflowRuns(ctx context.Context, db DBTX, arg ListWorkflo
&i.WorkflowRunTriggeredBy.Input,
&i.WorkflowRunTriggeredBy.ParentId,
&i.WorkflowRunTriggeredBy.CronName,
&i.WorkflowRunTriggeredBy.IdentityId,
&i.WorkflowVersion.ID,
&i.WorkflowVersion.CreatedAt,
&i.WorkflowVersion.UpdatedAt,
@@ -2771,7 +2752,7 @@ WHERE
"WorkflowRun".id = eligible_runs.id AND
"WorkflowRun"."status" = 'QUEUED'
RETURNING
"WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder", "WorkflowRun"."identityId"
"WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder"
`
type PopWorkflowRunsRoundRobinParams struct {
@@ -2810,7 +2791,6 @@ func (q *Queries) PopWorkflowRunsRoundRobin(ctx context.Context, db DBTX, arg Po
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -2838,7 +2818,7 @@ SET
"cancelledError" = NULL
WHERE
"id" = $1::uuid
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId", "identityId"
RETURNING id, "createdAt", "updatedAt", "deletedAt", "tenantId", "jobId", "tickerId", status, result, "startedAt", "finishedAt", "timeoutAt", "cancelledAt", "cancelledReason", "cancelledError", "workflowRunId"
`
func (q *Queries) ReplayWorkflowRunResetJobRun(ctx context.Context, db DBTX, jobrunid pgtype.UUID) (*JobRun, error) {
@@ -2861,7 +2841,6 @@ func (q *Queries) ReplayWorkflowRunResetJobRun(ctx context.Context, db DBTX, job
&i.CancelledReason,
&i.CancelledError,
&i.WorkflowRunId,
&i.IdentityId,
)
return &i, err
}
@@ -3072,7 +3051,7 @@ SET
WHERE
"tenantId" = $5::uuid AND
"id" = ANY($6::uuid[])
RETURNING "WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder", "WorkflowRun"."identityId"
RETURNING "WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder"
`
type UpdateManyWorkflowRunParams struct {
@@ -3121,7 +3100,6 @@ func (q *Queries) UpdateManyWorkflowRun(ctx context.Context, db DBTX, arg Update
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
); err != nil {
return nil, err
}
@@ -3168,7 +3146,7 @@ SET
WHERE
"id" = $5::uuid AND
"tenantId" = $6::uuid
RETURNING "WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder", "WorkflowRun"."identityId"
RETURNING "WorkflowRun"."createdAt", "WorkflowRun"."updatedAt", "WorkflowRun"."deletedAt", "WorkflowRun"."tenantId", "WorkflowRun"."workflowVersionId", "WorkflowRun".status, "WorkflowRun".error, "WorkflowRun"."startedAt", "WorkflowRun"."finishedAt", "WorkflowRun"."concurrencyGroupId", "WorkflowRun"."displayName", "WorkflowRun".id, "WorkflowRun"."childIndex", "WorkflowRun"."childKey", "WorkflowRun"."parentId", "WorkflowRun"."parentStepRunId", "WorkflowRun"."additionalMetadata", "WorkflowRun".duration, "WorkflowRun".priority, "WorkflowRun"."insertOrder"
`
type UpdateWorkflowRunParams struct {
@@ -3211,7 +3189,6 @@ func (q *Queries) UpdateWorkflowRun(ctx context.Context, db DBTX, arg UpdateWork
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}
@@ -3292,7 +3269,7 @@ FROM
WHERE
workflowRun."id" = groupKeyRun."workflowRunId" AND
workflowRun."tenantId" = $1::uuid
RETURNING workflowrun."createdAt", workflowrun."updatedAt", workflowrun."deletedAt", workflowrun."tenantId", workflowrun."workflowVersionId", workflowrun.status, workflowrun.error, workflowrun."startedAt", workflowrun."finishedAt", workflowrun."concurrencyGroupId", workflowrun."displayName", workflowrun.id, workflowrun."childIndex", workflowrun."childKey", workflowrun."parentId", workflowrun."parentStepRunId", workflowrun."additionalMetadata", workflowrun.duration, workflowrun.priority, workflowrun."insertOrder", workflowrun."identityId"
RETURNING workflowrun."createdAt", workflowrun."updatedAt", workflowrun."deletedAt", workflowrun."tenantId", workflowrun."workflowVersionId", workflowrun.status, workflowrun.error, workflowrun."startedAt", workflowrun."finishedAt", workflowrun."concurrencyGroupId", workflowrun."displayName", workflowrun.id, workflowrun."childIndex", workflowrun."childKey", workflowrun."parentId", workflowrun."parentStepRunId", workflowrun."additionalMetadata", workflowrun.duration, workflowrun.priority, workflowrun."insertOrder"
`
type UpdateWorkflowRunGroupKeyFromRunParams struct {
@@ -3324,7 +3301,6 @@ func (q *Queries) UpdateWorkflowRunGroupKeyFromRun(ctx context.Context, db DBTX,
&i.Duration,
&i.Priority,
&i.InsertOrder,
&i.IdentityId,
)
return &i, err
}

3
pkg/repository/prisma/dbsqlc/workflows.sql.go
View File

@@ -1863,7 +1863,7 @@ func (q *Queries) ListWorkflowsForEvent(ctx context.Context, db DBTX, arg ListWo
const listWorkflowsLatestRuns = `-- name: ListWorkflowsLatestRuns :many
SELECT
DISTINCT ON (workflow."id") runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder", runs."identityId", workflow."id" as "workflowId"
DISTINCT ON (workflow."id") runs."createdAt", runs."updatedAt", runs."deletedAt", runs."tenantId", runs."workflowVersionId", runs.status, runs.error, runs."startedAt", runs."finishedAt", runs."concurrencyGroupId", runs."displayName", runs.id, runs."childIndex", runs."childKey", runs."parentId", runs."parentStepRunId", runs."additionalMetadata", runs.duration, runs.priority, runs."insertOrder", workflow."id" as "workflowId"
FROM
"WorkflowRun" as runs
LEFT JOIN
@@ -1945,7 +1945,6 @@ func (q *Queries) ListWorkflowsLatestRuns(ctx context.Context, db DBTX, arg List
&i.WorkflowRun.Duration,
&i.WorkflowRun.Priority,
&i.WorkflowRun.InsertOrder,
&i.WorkflowRun.IdentityId,
&i.WorkflowId,
); err != nil {
return nil, err

260
sql/migrations/20241121142159_v0.52.0.sql
View File

@@ -1,244 +1,18 @@
-- atlas:txmode none
-- 'Creating an updatedAt index that will be useful later';
create index CONCURRENTLY IF NOT EXISTS "StepRun_updatedAt_idx" on "StepRun" ("updatedAt");
-- 'Created index';
DO $$
DECLARE
retry_count INT := 0;
max_retries INT := 10;
sleep_interval INT := 5000;
rec RECORD;
sql_statement TEXT;
newest_record RECORD;
BEGIN
WHILE retry_count < max_retries LOOP
BEGIN
SET LOCAL lock_timeout = '30s';
CREATE TABLE "StepRun_new" (
"id" uuid NOT NULL,
"createdAt" timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"deletedAt" timestamp(3),
"tenantId" uuid NOT NULL,
"jobRunId" uuid NOT NULL,
"stepId" uuid NOT NULL,
"order" BIGSERIAL,
"workerId" uuid,
"tickerId" uuid,
"status" "StepRunStatus" NOT NULL DEFAULT 'PENDING'::"StepRunStatus",
"input" jsonb,
"output" jsonb,
"requeueAfter" timestamp(3) without time zone,
"scheduleTimeoutAt" timestamp(3) without time zone,
"error" text,
"startedAt" timestamp(3) without time zone,
"finishedAt" timestamp(3) without time zone,
"timeoutAt" timestamp(3) without time zone,
"cancelledAt" timestamp(3) without time zone,
"cancelledReason" text,
"cancelledError" text,
"inputSchema" jsonb,
"callerFiles" jsonb,
"gitRepoBranch" text,
"retryCount" integer NOT NULL DEFAULT 0,
"semaphoreReleased" boolean NOT NULL DEFAULT false,
"queue" text NOT NULL DEFAULT 'default'::text,
"priority" integer,
"internalRetryCount" INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY ( "status", "id")
) PARTITION BY LIST ("status");
RAISE NOTICE 'Created table "StepRun_new"';
CREATE TABLE "StepRun_volatile" PARTITION OF "StepRun_new"
FOR VALUES IN ('PENDING','PENDING_ASSIGNMENT','ASSIGNED','RUNNING','CANCELLING')
WITH (fillfactor = 50);
CREATE TABLE "StepRun_stable" PARTITION OF "StepRun_new"
FOR VALUES IN ('FAILED','CANCELLED','SUCCEEDED') WITH (fillfactor = 100);
RAISE NOTICE 'Created partitions for "StepRun_new"';
RAISE NOTICE 'Inserting data into "StepRun_new"';
INSERT INTO "StepRun_new" SELECT * FROM "StepRun" ;
RAISE NOTICE 'Inserted data into "StepRun_new"';
RAISE NOTICE 'Inserted data into "StepRun_new"';
RAISE NOTICE 'Creating indexes for "StepRun_new"';
ALTER INDEX IF EXISTS "StepRun_createdAt_idx" RENAME TO "StepRun_createdAt_idx_old";
ALTER INDEX IF EXISTS "StepRun_deletedAt_idx" RENAME TO "StepRun_deletedAt_idx_old";
ALTER INDEX IF EXISTS "StepRun_updatedAt_idx" RENAME TO "StepRun_updatedAt_idx_old";
ALTER INDEX IF EXISTS "StepRun_id_tenantId_idx" RENAME TO "StepRun_id_tenantId_idx_old";
ALTER INDEX IF EXISTS "StepRun_jobRunId_status_idx" RENAME TO "StepRun_jobRunId_status_idx_old";
ALTER INDEX IF EXISTS "StepRun_jobRunId_tenantId_order_idx" RENAME TO "StepRun_jobRunId_tenantId_order_idx_old";
ALTER INDEX IF EXISTS "StepRun_stepId_idx" RENAME TO "StepRun_stepId_idx_old";
ALTER INDEX IF EXISTS "StepRun_tenantId_idx" RENAME TO "StepRun_tenantId_idx_old";
ALTER INDEX IF EXISTS "StepRun_workerId_idx" RENAME TO "StepRun_workerId_idx_old";
CREATE INDEX "StepRun_createdAt_idx" ON "StepRun_new" ("createdAt");
CREATE INDEX "StepRun_deletedAt_idx" ON "StepRun_new" ("deletedAt");
CREATE INDEX "StepRun_updatedAt_idx" ON "StepRun_new" ("updatedAt");
CREATE INDEX "StepRun_id_tenantId_idx" ON "StepRun_new" ("id", "tenantId");
CREATE INDEX "StepRun_jobRunId_status_idx" ON "StepRun_new" ("jobRunId", "status");
CREATE INDEX "StepRun_jobRunId_tenantId_order_idx" ON "StepRun_new" ("jobRunId", "tenantId", "order");
CREATE INDEX "StepRun_stepId_idx" ON "StepRun_new" ("stepId");
CREATE INDEX "StepRun_tenantId_idx" ON "StepRun_new" ("tenantId");
CREATE INDEX "StepRun_workerId_idx" ON "StepRun_new" ("workerId");
CREATE INDEX "StepRun_status_tenantId_idx" ON "StepRun_new" ("status", "tenantId");
RAISE NOTICE 'Created indexes for "StepRun_new"';
RAISE NOTICE 'Checking for data since the last select';
INSERT INTO "StepRun_new" SELECT * FROM "StepRun" where "updatedAt" >= (SELECT max("updatedAt") FROM "StepRun_new") AND NOT EXISTS (
SELECT 1 FROM "StepRun_new" WHERE "StepRun_new"."id" = "StepRun"."id"
);
ALTER TABLE "StepRun_volatile"
SET (autovacuum_vacuum_threshold = '1000',
autovacuum_vacuum_scale_factor = '0.01',
autovacuum_analyze_threshold = '500',
autovacuum_analyze_scale_factor = '0.01');
RAISE NOTICE 'Renaming tables and copying any new data';
BEGIN
LOCK TABLE "StepRun" IN SHARE MODE;
LOCK TABLE "StepRun_new" IN SHARE MODE;
INSERT INTO "StepRun_new"
SELECT *
FROM "StepRun"
WHERE "updatedAt" >= (SELECT max("updatedAt") FROM "StepRun_new")
AND NOT EXISTS (
SELECT 1 FROM "StepRun_new" WHERE "StepRun_new"."id" = "StepRun"."id"
);
ALTER TABLE "StepRun" RENAME TO "StepRun_old";
ALTER TABLE "StepRun_new" RENAME TO "StepRun";
END;
FOR rec IN
SELECT
conname AS constraint_name,
conrelid::regclass AS referencing_table,
a.attname AS referencing_column,
confrelid::regclass AS referenced_table,
af.attname AS referenced_column
FROM
pg_constraint c
JOIN
pg_attribute a ON a.attnum = ANY(c.conkey) AND a.attrelid = c.conrelid
JOIN
pg_attribute af ON af.attnum = ANY(c.confkey) AND af.attrelid = c.confrelid
WHERE
confrelid = '"StepRun_old"'::regclass
AND contype = 'f'
LOOP
RAISE NOTICE 'Referencing column: %, Referenced table: %, Referenced column: %', rec.referencing_column, rec.referenced_table, rec.referenced_column;
sql_statement = 'CREATE OR REPLACE FUNCTION ' || 'StepRun' || rec.referencing_column || '_fk_trigger_function() RETURNS TRIGGER AS $function_body$
BEGIN
IF NEW."'|| rec.referencing_column || '" IS NOT NULL THEN
IF NOT EXISTS (
SELECT 1
FROM "StepRun"
WHERE "' || rec.referenced_column || '" = NEW."' || rec.referencing_column || '"
) THEN
RAISE EXCEPTION ''Foreign key violation: ' || 'StepRun' || ' with ' || rec.referenced_column || ' = % does not exist.'', NEW."' || rec.referencing_column || '";
END IF;
END IF;
RETURN NEW;
END;
$function_body$ LANGUAGE plpgsql;';
RAISE NOTICE 'Executing: %', sql_statement;
EXECUTE sql_statement;
RAISE NOTICE 'Created trigger function for %', rec.constraint_name;
sql_statement = 'CREATE TRIGGER "' || rec.constraint_name || '_fk_insert_trigger"
BEFORE INSERT ON ' || rec.referencing_table || '
FOR EACH ROW
EXECUTE FUNCTION ' || 'StepRun'||rec.referencing_column|| '_fk_trigger_function();';
RAISE NOTICE 'Executing: %', sql_statement;
EXECUTE sql_statement;
sql_statement = 'CREATE TRIGGER "' || rec.constraint_name || '_fk_update_trigger"
BEFORE UPDATE ON ' || rec.referencing_table || '
FOR EACH ROW
EXECUTE FUNCTION ' || 'StepRun' || rec.referencing_column|| '_fk_trigger_function();';
RAISE NOTICE 'Executing: %', sql_statement;
EXECUTE sql_statement;
sql_statement = 'ALTER TABLE ' || rec.referencing_table || ' DROP CONSTRAINT "' || rec.constraint_name || '"';
RAISE NOTICE 'Executing: %', sql_statement;
EXECUTE sql_statement;
END LOOP;
RAISE NOTICE 'Migration successful EXIT';
EXIT;
EXCEPTION
WHEN OTHERS THEN
RAISE NOTICE 'Migration failed, retrying...';
RAISE NOTICE 'SQLSTATE: %, Message: %', SQLSTATE, SQLERRM;
ROLLBACK;
retry_count := retry_count + 1;
RAISE NOTICE 'Attempt %', retry_count;
PERFORM pg_sleep(sleep_interval / 1000.0);
END;
END LOOP;
IF retry_count = max_retries THEN
RAISE EXCEPTION 'Migration failed after % attempts.', max_retries;
END IF;
RAISE NOTICE 'Migration successful COMMIT';
DROP TABLE "StepRun_old";
COMMIT;
END $$;
-- Drop index "StepRun_updatedAt_idx" from table: "StepRun"
DROP INDEX "StepRun_updatedAt_idx";
-- Create index "StepRun_id_key" to table: "StepRun"
CREATE UNIQUE INDEX "StepRun_id_key" ON "StepRun" ("id", "status");
-- Create index "StepRun_jobRunId_status_tenantId_idx" to table: "StepRun"
CREATE INDEX "StepRun_jobRunId_status_tenantId_idx" ON "StepRun" ("jobRunId", "status", "tenantId") WHERE (status = 'PENDING'::"StepRunStatus");
-- Modify "StepRun" table
ALTER TABLE "StepRun" ADD COLUMN "identityId" bigserial NOT NULL, ADD CONSTRAINT "step_run_identity_id_unique" UNIQUE ("identityId", "status");
-- Modify "Event" table
ALTER TABLE "Event" ADD COLUMN "identityId" bigint NOT NULL GENERATED ALWAYS AS IDENTITY;
-- Modify "JobRun" table
ALTER TABLE "JobRun" ADD COLUMN "identityId" bigint NOT NULL GENERATED ALWAYS AS IDENTITY;
-- Modify "JobRunLookupData" table
ALTER TABLE "JobRunLookupData" ADD COLUMN "identityId" bigint NOT NULL GENERATED ALWAYS AS IDENTITY;
-- Modify "WorkflowRun" table
ALTER TABLE "WorkflowRun" ADD COLUMN "identityId" bigint NOT NULL GENERATED ALWAYS AS IDENTITY;
-- Modify "Step" table
ALTER TABLE "Step" ADD COLUMN "retryBackoffFactor" double precision NULL, ADD COLUMN "retryMaxBackoff" integer NULL;
-- Create "RetryQueueItem" table
CREATE TABLE "RetryQueueItem" ("id" bigserial NOT NULL, "retryAfter" timestamp(3) NOT NULL, "stepRunId" uuid NOT NULL, "tenantId" uuid NOT NULL, "isQueued" boolean NOT NULL, PRIMARY KEY ("id"));
-- Create index "RetryQueueItem_isQueued_tenantId_retryAfter_idx" to table: "RetryQueueItem"
CREATE INDEX "RetryQueueItem_isQueued_tenantId_retryAfter_idx" ON "RetryQueueItem" ("isQueued", "tenantId", "retryAfter");
-- Create enum type "WorkflowTriggerCronRefMethods"
CREATE TYPE "WorkflowTriggerCronRefMethods" AS ENUM ('DEFAULT', 'API');
-- Create enum type "WorkflowTriggerScheduledRefMethods"
CREATE TYPE "WorkflowTriggerScheduledRefMethods" AS ENUM ('DEFAULT', 'API');
-- Modify "WorkflowTriggerCronRef" table
ALTER TABLE "WorkflowTriggerCronRef" ADD COLUMN "name" text NULL, ADD COLUMN "id" uuid NOT NULL, ADD COLUMN "method" "WorkflowTriggerCronRefMethods" NOT NULL DEFAULT 'DEFAULT', ADD CONSTRAINT "WorkflowTriggerCronRef_parentId_cron_name_key" UNIQUE ("parentId", "cron", "name");
-- Modify "WorkflowTriggerScheduledRef" table
ALTER TABLE "WorkflowTriggerScheduledRef" ADD COLUMN "method" "WorkflowTriggerScheduledRefMethods" NOT NULL DEFAULT 'DEFAULT';
-- Modify "WorkflowRunTriggeredBy" table
ALTER TABLE "WorkflowRunTriggeredBy" ADD COLUMN "identityId" bigint NOT NULL GENERATED ALWAYS AS IDENTITY;
ALTER TABLE "WorkflowRunTriggeredBy" DROP CONSTRAINT "WorkflowRunTriggeredBy_cronParentId_cronSchedule_fkey", ADD COLUMN "cronName" text NULL, ADD CONSTRAINT "WorkflowRunTriggeredBy_cronParentId_cronSchedule_cronName_fkey" FOREIGN KEY ("cronParentId", "cronSchedule", "cronName") REFERENCES "WorkflowTriggerCronRef" ("parentId", "cron", "name") ON UPDATE CASCADE ON DELETE SET NULL;
-- Drop index "WorkflowTriggerCronRef_parentId_cron_key" from table: "WorkflowTriggerCronRef"
DROP INDEX "WorkflowTriggerCronRef_parentId_cron_key";

6
sql/migrations/20241121153627_v0.52.1.sql
View File

@@ -1,6 +0,0 @@
-- Modify "Step" table
ALTER TABLE "Step" ADD COLUMN "retryBackoffFactor" double precision NULL, ADD COLUMN "retryMaxBackoff" integer NULL;
-- Create "RetryQueueItem" table
CREATE TABLE "RetryQueueItem" ("id" bigserial NOT NULL, "retryAfter" timestamp(3) NOT NULL, "stepRunId" uuid NOT NULL, "tenantId" uuid NOT NULL, "isQueued" boolean NOT NULL, PRIMARY KEY ("id"));
-- Create index "RetryQueueItem_isQueued_tenantId_retryAfter_idx" to table: "RetryQueueItem"
CREATE INDEX "RetryQueueItem_isQueued_tenantId_retryAfter_idx" ON "RetryQueueItem" ("isQueued", "tenantId", "retryAfter");

12
sql/migrations/20241121195232_v0.52.2.sql
View File

@@ -1,12 +0,0 @@
-- Create enum type "WorkflowTriggerCronRefMethods"
CREATE TYPE "WorkflowTriggerCronRefMethods" AS ENUM ('DEFAULT', 'API');
-- Create enum type "WorkflowTriggerScheduledRefMethods"
CREATE TYPE "WorkflowTriggerScheduledRefMethods" AS ENUM ('DEFAULT', 'API');
-- Modify "WorkflowTriggerCronRef" table
ALTER TABLE "WorkflowTriggerCronRef" ADD COLUMN "name" text NULL, ADD COLUMN "id" uuid NOT NULL, ADD COLUMN "method" "WorkflowTriggerCronRefMethods" NOT NULL DEFAULT 'DEFAULT', ADD CONSTRAINT "WorkflowTriggerCronRef_parentId_cron_name_key" UNIQUE ("parentId", "cron", "name");
-- Modify "WorkflowTriggerScheduledRef" table
ALTER TABLE "WorkflowTriggerScheduledRef" ADD COLUMN "method" "WorkflowTriggerScheduledRefMethods" NOT NULL DEFAULT 'DEFAULT';
-- Modify "WorkflowRunTriggeredBy" table
ALTER TABLE "WorkflowRunTriggeredBy" DROP CONSTRAINT "WorkflowRunTriggeredBy_cronParentId_cronSchedule_fkey", ADD COLUMN "cronName" text NULL, ADD CONSTRAINT "WorkflowRunTriggeredBy_cronParentId_cronSchedule_cronName_fkey" FOREIGN KEY ("cronParentId", "cronSchedule", "cronName") REFERENCES "WorkflowTriggerCronRef" ("parentId", "cron", "name") ON UPDATE CASCADE ON DELETE SET NULL;
-- Drop index "WorkflowTriggerCronRef_parentId_cron_key" from table: "WorkflowTriggerCronRef"
DROP INDEX "WorkflowTriggerCronRef_parentId_cron_key";

6
sql/migrations/atlas.sum
View File

@@ -1,4 +1,4 @@
h1:dC6PwKKRlUBxyRThtmBxRZoUsxdOH3dTos0926Yij54=
h1:ttKQZ/+X6AnYVp7nozbGyoby/BNrbML2MxWaMwrgNmA=
20240115180414_init.sql h1:Ef3ZyjAHkmJPdGF/dEWCahbwgcg6uGJKnDxW2JCRi2k=
20240122014727_v0_6_0.sql h1:o/LdlteAeFgoHJ3e/M4Xnghqt9826IE/Y/h0q95Acuo=
20240126235456_v0_7_0.sql h1:KiVzt/hXgQ6esbdC6OMJOOWuYEXmy1yeCpmsVAHTFKs=
@@ -75,6 +75,4 @@ h1:dC6PwKKRlUBxyRThtmBxRZoUsxdOH3dTos0926Yij54=
20241029122625_v0.51.0.sql h1:nOa4FqmZxSh1yBOJyduX+j15gQavjizTn660wyXjhNk=
20241107162939_v0.51.2.sql h1:qtnUITelb0kzAazo99gdTzejmQeOiE8NTP8b8bpQuF0=
20241114175346_v0.51.3.sql h1:ZbpRJsCmt6098ilZ3LtOk9LXRzuuwiznXPJmSkZSRpg=
20241121142159_v0.52.0.sql h1:JGnuMOg3QTs2u9mcgMGA1PNqjrA64zyHpFVm9IRCkAg=
20241121153627_v0.52.1.sql h1:6ZHMeSYHjo1MRS1qDJH6RzNKdR46YQzdJ9DxCm56IlM=
20241121195232_v0.52.2.sql h1:ExHYa+nRVVff5Gb1dvRwHyC98eFRVA0Vsrg/bd96hYE=
20241121142159_v0.52.0.sql h1:nVceeHB+PW9R0FBTCalGamY1l9n5mOrmCpuiH1QkVB4=

109
sql/schema/schema.sql
View File

@@ -192,7 +192,6 @@ CREATE TABLE "Event" (
"data" JSONB,
"additionalMetadata" JSONB,
"insertOrder" INTEGER,
"identityId" BIGINT GENERATED ALWAYS AS IDENTITY,
CONSTRAINT "Event_pkey" PRIMARY KEY ("id")
);
@@ -279,7 +278,6 @@ CREATE TABLE "JobRun" (
"cancelledReason" TEXT,
"cancelledError" TEXT,
"workflowRunId" UUID NOT NULL,
"identityId" BIGINT GENERATED ALWAYS AS IDENTITY,
CONSTRAINT "JobRun_pkey" PRIMARY KEY ("id")
);
@@ -293,7 +291,6 @@ CREATE TABLE "JobRunLookupData" (
"jobRunId" UUID NOT NULL,
"tenantId" UUID NOT NULL,
"data" JSONB,
"identityId" BIGINT GENERATED ALWAYS AS IDENTITY,
CONSTRAINT "JobRunLookupData_pkey" PRIMARY KEY ("id")
);
@@ -486,78 +483,40 @@ CREATE TABLE "StepRateLimit" (
CREATE SEQUENCE steprun_identity_id_seq START 1;
-- CreateTable
CREATE TABLE
"StepRun" (
"id" UUID NOT NULL,
"createdAt" TIMESTAMP(3) without time zone NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) without time zone NOT NULL DEFAULT CURRENT_TIMESTAMP,
"deletedAt" TIMESTAMP(3) without time zone,
"tenantId" UUID NOT NULL,
"jobRunId" UUID NOT NULL,
"stepId" UUID NOT NULL,
"order" BIGSERIAL NOT NULL,
"workerId" UUID,
"tickerId" UUID,
"status" "StepRunStatus" NOT NULL DEFAULT 'PENDING',
"input" JSONB,
"output" JSONB,
"requeueAfter" TIMESTAMP(3),
"scheduleTimeoutAt" TIMESTAMP(3),
"error" TEXT,
"startedAt" TIMESTAMP(3),
"finishedAt" TIMESTAMP(3),
"timeoutAt" TIMESTAMP(3),
"cancelledAt" TIMESTAMP(3),
"cancelledReason" TEXT,
"cancelledError" TEXT,
"inputSchema" JSONB,
"callerFiles" JSONB,
"gitRepoBranch" TEXT,
"retryCount" INTEGER NOT NULL DEFAULT 0,
"semaphoreReleased" BOOLEAN NOT NULL DEFAULT false,
"queue" TEXT NOT NULL DEFAULT 'default',
"priority" INTEGER,
"internalRetryCount" INTEGER NOT NULL DEFAULT 0,
"identityId" BIGINT NOT NULL DEFAULT nextval('steprun_identity_id_seq'::regclass),
CONSTRAINT "StepRun_pkey" PRIMARY KEY ("status", "id")
)
PARTITION BY
LIST ("status");
ALTER TABLE "StepRun"
ADD CONSTRAINT step_run_identity_id_unique UNIQUE ("identityId","status");
CREATE TABLE
"StepRun_volatile" PARTITION OF "StepRun" FOR
VALUES
IN (
'PENDING',
'PENDING_ASSIGNMENT',
'ASSIGNED',
'RUNNING',
'CANCELLING'
)
WITH
(fillfactor = 50);
ALTER TABLE "StepRun_volatile"
SET
(
autovacuum_vacuum_threshold = '1000',
autovacuum_vacuum_scale_factor = '0.01',
autovacuum_analyze_threshold = '500',
autovacuum_analyze_scale_factor = '0.01'
);
CREATE TABLE
"StepRun_stable" PARTITION OF "StepRun" FOR
VALUES
IN ('FAILED', 'CANCELLED', 'SUCCEEDED')
WITH
(fillfactor = 100);
CREATE TABLE "StepRun" (
"id" UUID NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"deletedAt" TIMESTAMP(3),
"tenantId" UUID NOT NULL,
"jobRunId" UUID NOT NULL,
"stepId" UUID NOT NULL,
"order" BIGSERIAL NOT NULL,
"workerId" UUID,
"tickerId" UUID,
"status" "StepRunStatus" NOT NULL DEFAULT 'PENDING',
"input" JSONB,
"output" JSONB,
"requeueAfter" TIMESTAMP(3),
"scheduleTimeoutAt" TIMESTAMP(3),
"error" TEXT,
"startedAt" TIMESTAMP(3),
"finishedAt" TIMESTAMP(3),
"timeoutAt" TIMESTAMP(3),
"cancelledAt" TIMESTAMP(3),
"cancelledReason" TEXT,
"cancelledError" TEXT,
"inputSchema" JSONB,
"callerFiles" JSONB,
"gitRepoBranch" TEXT,
"retryCount" INTEGER NOT NULL DEFAULT 0,
"semaphoreReleased" BOOLEAN NOT NULL DEFAULT false,
"queue" TEXT NOT NULL DEFAULT 'default',
"priority" INTEGER,
"internalRetryCount" INTEGER NOT NULL DEFAULT 0,
CONSTRAINT "StepRun_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "StepRunEvent" (
@@ -951,7 +910,6 @@ CREATE TABLE "WorkflowRun" (
"duration" BIGINT,
"priority" INTEGER,
"insertOrder" INTEGER,
"identityId" BIGINT GENERATED ALWAYS AS IDENTITY,
CONSTRAINT "WorkflowRun_pkey" PRIMARY KEY ("id")
);
@@ -994,7 +952,6 @@ CREATE TABLE "WorkflowRunTriggeredBy" (
"input" JSONB,
"parentId" UUID NOT NULL,
"cronName" TEXT,
"identityId" BIGINT GENERATED ALWAYS AS IDENTITY,
CONSTRAINT "WorkflowRunTriggeredBy_pkey" PRIMARY KEY ("id")
);