Unfortunaly, aggregate plugin can't help you as it works today.
It needs a clearly identified "task end event" to push aggregate map into end task event.
In that case, we can clearly detect "task start event" : it is the first one with id=1.
To fill the need, I plan to add a new option in aggregate plugin :
push_previous_map_as_event => true
When this option is activated, each time a new task is detected (in your case, each time a new id is detected), it automatically push previous aggregate map as a new event in logstash.
Using that feature in your case, you can enrich aggregate map with each db line result, and when aggregate detects a new id, it push map with all aggregated data info as new event.