379 round robin pipe iterator #421
Conversation
✅ Deploy Preview for dlt-hub-docs ready!
To edit notification comments on pull requests, go to your Netlify site settings. |
There was a problem hiding this comment.
there's one important things in the ticket #379
that the newly added sources (gens) must be ALWAYS evaluated before those added by intialization.
execution of the step may create a generator. to preserve any deadlocks we evaluate the newly added generators without the round robin. see the ticket.
dlt/extract/pipe.py
Outdated
| def _get_source_item_round_robin(self) -> ResolvablePipeItem: | ||
| try: | ||
| # get items from last added iterator, this makes the overall Pipe as close to FIFO as possible | ||
| self._round_robin_index = self._round_robin_index % len(self._sources) |
There was a problem hiding this comment.
I'd start the index at -1 and increase it by 1 here then takin modulo in the next line. so we save on assignment. I know I'm pedantic but this is the most frequently called piece of code
| self._round_robin_index -= 1 | ||
| # remove empty iterator and try another source | ||
| self._sources.pop(self._round_robin_index) | ||
| return self._get_source_item() |
There was a problem hiding this comment.
not the round robin again?
There was a problem hiding this comment.
I changed the code a bit, and it has to stay like this for round robin and current to work in conjunction
dlt/extract/pipe.py
Outdated
| @@ -670,13 +676,18 @@ def _get_source_item(self) -> ResolvablePipeItem: | |||
| # no more sources to iterate | |||
| if len(self._sources) == 0: | |||
| return None | |||
| if self._next_item_mode == "current": | |||
| return self._get_source_item_current() | |||
| if self._next_item_mode == "round_robin": | |||
There was a problem hiding this comment.
I'd do elif because it is faster
| @@ -15,6 +15,35 @@ | |||
| # from tests.utils import preserve_environ | |||
|
|
|||
|
|
|||
| def test_next_item_mode() -> None: | |||
There was a problem hiding this comment.
it's even better now :D
sh-rp
force-pushed
the
d#/379-round-robin-pipe-iterator
branch
from
e038023
to
577c244
Compare
| yield from [1, 2, nested_gen(), 3,4] | ||
|
|
||
| def source_gen2(): | ||
| yield from range(11,16) |
There was a problem hiding this comment.
just intuition: could you add nested gen at the end of the range?
There was a problem hiding this comment.
seems to work without a problem
tests/extract/test_extract_pipe.py
Outdated
|
|
||
| # test mode "round robin" | ||
| reversed_pipes = get_pipes() | ||
| reversed_pipes.reverse() # reverse order of pipes for easier testing |
There was a problem hiding this comment.
good point! this actually must be reversed in the code, not in the test. the first pipe in the list must be evaluated first! I think I reverse the them inside so skip the reverse for round robin
implements #379