code: func collectStepOutputDatasetSize(status *pb.FlowExecutionStatus, stepId int32) (counter int64) {
  	for _, tg := range status.TaskGroups {
  		for _, execution := range tg.Executions {
  			if execution.ExecutionStat == nil {
  				continue
  			}
  			for _, stat := range execution.ExecutionStat.Stats {
  				if stat.StepId == stepId {
  					counter += stat.OutputCounter
  				}
  			}
  		}
  	}
  	return counter
  }
docs: // may be more efficient to map stepId=>size