Repository: quantopian/zipline
File: zipline/pipeline/engine.py (Python)

Code before:
                self._root_mask_term: root_mask_values,
                self._root_mask_dates_term: as_column(dates.values)
            },
            self._root_mask_term,
            graph,
            dates,
            assets,
        )

        results = self.compute_chunk(
            graph,
            dates,
            assets,
            initial_workspace,
        )

        return self._to_narrow(
            graph.outputs,
            results,
            results.pop(screen_name),
            dates[extra_rows:],
            assets,
        )

    def run_chunked_pipeline(self, pipeline, start_date, end_date, chunksize):
        """Run a pipeline to collect the results.

        Parameters
        ----------
        pipeline : Pipeline
            The pipeline to run.
        start_date : pd.Timestamp
            The start date to run the pipeline for.
        end_date : pd.Timestamp
            The end date to run the pipeline for.
        chunksize : int or None
            The number of days to execute at a time. If this is None, the
            entire date range will be run at once.

        Returns
        -------
        results : pd.DataFrame
            The results for each output term in the pipeline.
        """
        ranges = compute_date_range_chunks(
            self._calendar,
            start_date,
            end_date,
            chunksize,
        )
        chunks = [self.run_pipeline(pipeline, s, e) for s, e in ranges]

Review comment (suggestion):
I'm not sure what "to collect the results" means here.  Is there a reason to make this sentence different from the one in the interface definition?  Honestly we could probably just copy the docstring over from the base class using something like the `copydoc` function we use downsteam:

```python
# Taken from odo.utils.
@toolz.curry
def copydoc(from_, to):
    """Copies the docstring from one function to another.

    Parameters
    ----------
    from_ : any
        The object to copy the docstring from.
    to : any
        The object to copy the docstring to.

    Returns
    -------
    to : any
        ``to`` with the docstring from ``from_``
    """
    to.__doc__ = from_.__doc__
    return to
```

Code after:
            self._calendar,
            start_date,
            end_date,
        )
        extra_rows = graph.extra_rows[self._root_mask_term]
        root_mask = self._compute_root_mask(start_date, end_date, extra_rows)
        dates, assets, root_mask_values = explode(root_mask)

        initial_workspace = self._populate_initial_workspace(
            {
                self._root_mask_term: root_mask_values,
                self._root_mask_dates_term: as_column(dates.values)
            },
            self._root_mask_term,
            graph,
            dates,
            assets,
        )

        results = self.compute_chunk(
            graph,
            dates,
            assets,
            initial_workspace,
        )

        return self._to_narrow(
            graph.outputs,
            results,
            results.pop(screen_name),
            dates[extra_rows:],
            assets,
        )

    @copydoc(PipelineEngine.run_chunked_pipeline)
    def run_chunked_pipeline(self, pipeline, start_date, end_date, chunksize):
        ranges = compute_date_range_chunks(
            self._calendar,
            start_date,
            end_date,
            chunksize,
        )
        chunks = [self.run_pipeline(pipeline, s, e) for s, e in ranges]

        return categorical_df_concat(chunks, inplace=True)

    def _compute_root_mask(self, start_date, end_date, extra_rows):
        """
        Compute a lifetimes matrix from our AssetFinder, then drop columns that
        didn't exist at all during the query dates.