def METHOD_NAME(pipeline_run: PipelineRun, logger: DictLogger) -> List[BlockRun]:
    from mage_integrations.sources.constants import BATCH_FETCH_LIMIT
    integration_pipeline = IntegrationPipeline.get(pipeline_run.pipeline_uuid)
    blocks = integration_pipeline.get_executable_blocks()
    executable_blocks = []
    # we expect that blocks in integration pipelines only have 1 downstream block, and
    # that there is only 1 root block and 1 leaf block.
    current_block = integration_pipeline.data_loader
    while True:
        block = find(lambda b: b.uuid == current_block.uuid, blocks)
        executable_blocks.append(block)
        downstream_blocks = current_block.downstream_blocks
        if len(downstream_blocks) == 0:
            break
        current_block = downstream_blocks[0]
    data_loader_block = integration_pipeline.data_loader
    data_exporter_block = integration_pipeline.data_exporter
    if data_exporter_block is None:
        raise Exception("No Destination Block was found. \
                         Mage expects 1 Source and 1 Destination Block")
    transformer_blocks = [b for b in executable_blocks if b.uuid not in [
        data_loader_block.uuid,
        data_exporter_block.uuid,
    ]]
    is_sql_source = integration_pipeline.source_uuid in SQL_SOURCES_UUID
    record_counts_by_stream = {}
    if is_sql_source:
        record_counts_by_stream = index_by(lambda x: x['id'], integration_pipeline.count_records())
    arr = []
    for stream in integration_pipeline.streams():
        tap_stream_id = stream['tap_stream_id']
        tags = dict(
            source=integration_pipeline.source_uuid,
            stream=tap_stream_id,
        )
        record_counts = None
        if is_sql_source:
            record_counts = record_counts_by_stream[tap_stream_id]['count']
        number_of_batches = math.ceil((record_counts or 1) / BATCH_FETCH_LIMIT)
        tags2 = merge_dict(tags, dict(
            number_of_batches=number_of_batches,
            record_counts=record_counts,
        ))
        logger.info(
            f"Number of records for stream {tap_stream_id}: "
            f"{'N/A' if record_counts is None else record_counts}.",
            tags=tags2,
        )
        logger.info(
            f"Number of batches for loading data from stream {tap_stream_id}: {number_of_batches}.",
            tags=tags2,
        )
        for idx in range(number_of_batches):
            arr += [
                pipeline_run.create_block_run(
                    f'{data_loader_block.uuid}:{tap_stream_id}:{idx}',
                    commit=False,
                ),
            ] + [pipeline_run.create_block_run(
                    f'{b.uuid}:{tap_stream_id}:{idx}',
                    commit=False,
                ) for b in transformer_blocks] + [
                pipeline_run.create_block_run(
                    f'{data_exporter_block.uuid}:{tap_stream_id}:{idx}',
                    commit=False,
                ),
            ]
            try:
                db_connection.session.commit()
            except Exception as e:
                db_connection.session.rollback()
                raise Exception(f'Failed to create block runs for {tap_stream_id}:{idx}.') from e
    return arr