(coll)
| 517 | blocker = checkpoint(*collections, split_every=split_every) |
| 518 | |
| 519 | def block_one(coll): |
| 520 | tok = tokenize(coll, blocker) |
| 521 | dsks = [] |
| 522 | rename = {} |
| 523 | for prev_name in get_collection_names(coll): |
| 524 | new_name = "wait_on-" + tokenize(prev_name, tok) |
| 525 | rename[prev_name] = new_name |
| 526 | layer = _build_map_layer( |
| 527 | chunks.bind, prev_name, new_name, coll, dependencies=(blocker,) |
| 528 | ) |
| 529 | dsks.append( |
| 530 | HighLevelGraph.from_collections( |
| 531 | new_name, layer, dependencies=(coll, blocker) |
| 532 | ) |
| 533 | ) |
| 534 | dsk = HighLevelGraph.merge(*dsks) |
| 535 | rebuild, args = coll.__dask_postpersist__() |
| 536 | return rebuild(dsk, *args, rename=rename) |
| 537 | |
| 538 | unpacked, repack = unpack_collections(*collections) |
| 539 | out = repack([block_one(coll) for coll in unpacked]) |
no test coverage detected
searching dependent graphs…