| 619 | return data_finished_blobs |
| 620 | |
| 621 | def new_reader(self, **kwargs): |
| 622 | readers = [] |
| 623 | for reader_builder in self._reader_builders: |
| 624 | reader = reader_builder.new_reader(**kwargs) |
| 625 | if isinstance(reader, Reader): |
| 626 | pass |
| 627 | elif hasattr(reader, 'reader'): |
| 628 | reader = reader.reader() |
| 629 | else: |
| 630 | raise ValueError('reader must be an instance of Reader or Pipe') |
| 631 | readers.append(reader) |
| 632 | |
| 633 | multi_reader = CompositeReader(self._names, readers) |
| 634 | assert multi_reader.schema() == self._schema |
| 635 | return multi_reader |