Repository: microsoft/graphrag
File: graphrag/index/run/workflow.py (Python)

Code before:
async def _inject_workflow_data_dependencies(
    workflow: Workflow,
    workflow_dependencies: dict[str, list[str]],
    dataset: pd.DataFrame,
    storage: PipelineStorage,
) -> None:
    """Inject the data dependencies into the workflow."""
    workflow.add_table(DEFAULT_INPUT_NAME, dataset)
    deps = workflow_dependencies[workflow.name]
    log.info("dependencies for %s: %s", workflow.name, deps)
    for id in deps:
        workflow_id = f"workflow:{id}"
        table = await _load_table_from_storage(f"{id}.parquet", storage)
        workflow.add_table(workflow_id, table)


async def _emit_workflow_output(
    workflow: Workflow, emitters: list[TableEmitter]
) -> pd.DataFrame:
    """Emit the workflow output."""
    output = cast(pd.DataFrame, workflow.output())
    rows, cols = output.shape
    # only write the final output if it has content
    # this is expressly designed to allow us to create
    # workflows with side effects that don't produce a formal output to save
    if rows > 0 and cols > 0:
        for emitter in emitters:
            await emitter.emit(workflow.name, output)
    return output


def _create_callback_chain(
    callbacks: WorkflowCallbacks | None, progress: ProgressReporter | None
) -> WorkflowCallbacks:
    """Create a callbacks manager."""
    manager = WorkflowCallbacksManager()
    if callbacks is not None:
        manager.register(callbacks)
    if progress is not None:
        manager.register(ProgressWorkflowCallbacks(progress))
    return manager


async def _process_workflow(
    workflow: Workflow,
    context: PipelineRunContext,
    callbacks: WorkflowCallbacks,
    emitters: list[TableEmitter],
    workflow_dependencies: dict[str, list[str]],
    dataset: pd.DataFrame,
    start_time: float,

Review comment (bug):
What about `if not output.empty`? The [docs](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.empty.html) suggest this better accounts for the edge case where there may be columns but no rows in a dataframe.

Code after:
async def _inject_workflow_data_dependencies(
    workflow: Workflow,
    workflow_dependencies: dict[str, list[str]],
    dataset: pd.DataFrame,
    storage: PipelineStorage,
) -> None:
    """Inject the data dependencies into the workflow."""
    workflow.add_table(DEFAULT_INPUT_NAME, dataset)
    deps = workflow_dependencies[workflow.name]
    log.info("dependencies for %s: %s", workflow.name, deps)
    for id in deps:
        workflow_id = f"workflow:{id}"
        table = await _load_table_from_storage(f"{id}.parquet", storage)
        workflow.add_table(workflow_id, table)


async def _emit_workflow_output(
    workflow: Workflow, emitters: list[TableEmitter]
) -> pd.DataFrame:
    """Emit the workflow output."""
    output = cast(pd.DataFrame, workflow.output())
    # only write the final output if it has content
    # this is expressly designed to allow us to create
    # workflows with side effects that don't produce a formal output to save
    if not output.empty:
        for emitter in emitters:
            await emitter.emit(workflow.name, output)
    return output


def _create_callback_chain(
    callbacks: WorkflowCallbacks | None, progress: ProgressReporter | None
) -> WorkflowCallbacks:
    """Create a callbacks manager."""
    manager = WorkflowCallbacksManager()
    if callbacks is not None:
        manager.register(callbacks)
    if progress is not None:
        manager.register(ProgressWorkflowCallbacks(progress))
    return manager


async def _process_workflow(
    workflow: Workflow,
    context: PipelineRunContext,
    callbacks: WorkflowCallbacks,
    emitters: list[TableEmitter],
    workflow_dependencies: dict[str, list[str]],
    dataset: pd.DataFrame,
    start_time: float,
    is_resume_run: bool,