message Task {
  string sheet_sha256 = 4;
  string schema_version = 10;
  TaskReleaseSource task_release_source = 13;
  // ... other fields
}

message Task {
  oneof source {
    string sheet_sha256 = 10;  // For regular (non-release) tasks
    string release = 13;       // For release-based tasks: projects/{project}/releases/{release}
  }
  // ... other fields
  // Removed: schema_version (field 10, replaced by sheet_sha256)
  // Removed: task_release_source (field 13, replaced by release)
}

// Creates N×M tasks
for _, database := range databases {
    for _, file := range release.Payload.Files {
        // Filter by applied versions
        if alreadyApplied(file.Version) {
            continue
        }
        // Create task(database, file)
        task := &store.TaskMessage{
            Payload: &storepb.Task{
                SheetSha256: file.SheetSha256,
                SchemaVersion: file.Version,
                TaskReleaseSource: &storepb.TaskReleaseSource{
                    File: formatReleaseFile(file.Id),
                },
            },
        }
        taskCreates = append(taskCreates, task)
    }
}

// Creates N tasks
for _, database := range databases {
    // Create task(database, release)
    task := &store.TaskMessage{
        InstanceID:   database.InstanceID,
        DatabaseName: &database.DatabaseName,
        Environment:  database.EffectiveEnvironmentID,
        Type:         storepb.Task_DATABASE_MIGRATE,
        Payload: &storepb.Task{
            SpecId:  spec.Id,
            Release: c.Release,  // Store release name, not individual files
            // Remove SheetSha256, SchemaVersion, TaskReleaseSource
        },
    }
    taskCreates = append(taskCreates, task)
}

func (exec *DatabaseMigrateExecutor) RunOnce(ctx context.Context, ...) {
    sheet := getSheet(task.Payload.GetSheetSha256())
    runMigration(ctx, ..., sheet, task.Payload.GetSchemaVersion())
}

func (exec *DatabaseMigrateExecutor) RunOnce(ctx context.Context, ...) {
    // Check if this is a release-based task
    if releaseName := task.Payload.GetRelease(); releaseName != "" {
        return exec.runReleaseTask(ctx, task, taskRunUID, releaseName)
    }

    // Fall back to single-sheet execution
    sheet := getSheet(task.Payload.GetSheetSha256())
    runMigration(ctx, ..., sheet, "")
}

func (exec *DatabaseMigrateExecutor) runReleaseTask(
    ctx context.Context,
    task *store.TaskMessage,
    taskRunUID int,
    releaseName string,
) error {
    // 1. Fetch release
    _, releaseUID, err := common.GetProjectReleaseUID(releaseName)
    release, err := exec.store.GetReleaseByUID(ctx, releaseUID)

    // 2. Get existing revisions for this database
    revisions, err := exec.store.ListRevisions(ctx, &store.FindRevisionMessage{
        InstanceID:   &task.InstanceID,
        DatabaseName: task.DatabaseName,
    })

    // 3. Build map of applied versions
    appliedVersions := make(map[string]bool)
    for _, revision := range revisions {
        if revision.Payload.Type == storepb.SchemaChangeType_VERSIONED {
            appliedVersions[revision.Version] = true
        }
    }

    // 4. Execute unapplied files in order
    for _, file := range release.Payload.Files {
        if file.Type != storepb.SchemaChangeType_VERSIONED {
            continue  // Skip declarative for now
        }

        // Skip if already applied
        if appliedVersions[file.Version] {
            continue
        }

        // Fetch sheet and execute
        sheet, err := exec.store.GetSheet(ctx, &store.FindSheetMessage{
            Sha256: &file.SheetSha256,
        })

        // Run this file's migration
        _, result, err := runMigration(
            ctx,
            ...,
            sheet,
            file.Version,  // Pass version from file, not task
        )
        if err != nil {
            return err  // Stop on first failure
        }

        // Migration succeeded, revision already recorded by runMigration
    }

    return nil
}

-- Update task payload JSONB structure
-- Remove schema_version and task_release_source fields
-- Existing tasks remain unchanged (backward compatible at DB level)
-- New tasks use the new structure

-- No migration needed since:
-- 1. We're reusing existing field numbers in proto
-- 2. JSONB is flexible - old tasks keep old structure
-- 3. New tasks will have new structure
-- 4. Execution logic checks which fields are present

type TaskPatch struct {
    // ... existing fields

    // Remove or deprecate:
    // SchemaVersion *string

    // Add:
    Release *string  // For setting release name on task
}

// Remove:
// if v := patch.SchemaVersion; v != nil {
//     payloadParts.Join(" || ", "jsonb_build_object('schemaVersion', ?::TEXT)", *v)
// }

// Add:
if v := patch.Release; v != nil {
    payloadParts.Join(" || ", "jsonb_build_object('release', ?::TEXT)", *v)
}

// Before:
DatabaseUpdate: &v1pb.Task_DatabaseUpdate{
    Sheet:              common.FormatSheet(project.ResourceID, task.Payload.GetSheetSha256()),
    SchemaVersion:      task.Payload.GetSchemaVersion(),
    DatabaseChangeType: databaseChangeType,
}

// After:
databaseUpdate := &v1pb.Task_DatabaseUpdate{
    DatabaseChangeType: databaseChangeType,
}

// Set either sheet or release
if releaseName := task.Payload.GetRelease(); releaseName != "" {
    databaseUpdate.Release = releaseName
} else {
    databaseUpdate.Sheet = common.FormatSheet(project.ResourceID, task.Payload.GetSheetSha256())
}

DatabaseUpdate: databaseUpdate