then read file and

@extract_columns(*["a","b"])

@Seth Stokes I think you might have found a 🐛 . Why don’t you add an input parameter to

backfilled_df

? E.g.

def backfilled_df(dummy_input: str) -> pd.DataFrame:
    backfilled_df = pd.read_csv()
    print(len(backfilled_df))
    return backfilled_df

since I think with an input parameter it behaves as intended

but without , you’re right, it seems to get re-executed… actually maybe not … struggling to recreate it now.

just didnt show in the pseudo-code, appologies

I got nothing

what executor are you using?

Should hopefully be an easyish fix

OK, I’m debugging this. Weird behavior. Have it so that… • half of the time it’s running twice • half of the time its running 3 times

Here’s my repro with a bit of information/debugging stuff injected:

import collections
import functools

from hamilton.htypes import Parallelizable, Collect

_fn_call_counter = collections.Counter()


def _track_fn_call(fn) -> callable:
    @functools.wraps(fn)
    def wrapped(*args, **kwargs):
        _fn_call_counter[fn.__name__] += 1
        return fn(*args, **kwargs)

    return wrapped


def _reset_counter():
    _fn_call_counter.clear()


@_track_fn_call
def not_to_repeat() -> int:
    print("not_to_repeat")
    return -1


@_track_fn_call
def number_to_repeat(iterations: int) -> Parallelizable[int]:
    for i in range(iterations):
        yield i


@_track_fn_call
def something_else_not_to_repeat() -> int:
    print("something_else_not_to_repeat")
    return -2


@_track_fn_call
def double(number_to_repeat: int) -> int:
    print(number_to_repeat)
    return number_to_repeat * 2


@_track_fn_call
def summed(double: Collect[int], not_to_repeat: int, something_else_not_to_repeat: int) -> int:
    return sum(double) + not_to_repeat + something_else_not_to_repeat

dr = (
        driver.Builder()
        .with_modules(parallel_collect_multiple_arguments)
        .enable_dynamic_execution(allow_experimental_mode=True)
        .with_remote_executor(SynchronousLocalTaskExecutor())
        .with_grouping_strategy(GroupByRepeatableBlocks())
        .build()
    )
    parallel_collect_multiple_arguments._reset_counter()
    res = dr.execute(["summed"], inputs={"iterations" : 10})
    assert res["summed"] == 45*2-2-1
    counts = parallel_collect_multiple_arguments._fn_call_counter
    assert counts == {
        "not_to_repeat": 1,
        "number_to_repeat": 1,
        "something_else_not_to_repeat": 1,
        "double": 10,
        "summed": 1,
    }

We have a fix! Grouped in with a few others as separate commits. One line change (although it doesn’t look it as I’ve added tests/done a little bit of related cleanup): https://github.com/DAGWorks-Inc/hamilton/pull/545/commits/79127d44dc7dee9af75b32b2be11cf612df224a4. The TL;DR is we track the state of each task. The tasks go from “INITIALIZED” when they’re created to “QUEUED” when all dependencies are completed. We’d put them on the queue initially and forgot to mark them as QUEUED, which made it so the state tracker decided to run them again. We could probably clean this up by a multi-queue system without external state tracking, but I think this fix is pretty simple and there are some architectural trade-offs that I haven’t scoped out yet. That said @Seth Stokes I’d be surprised if this is responsible for the snlowness you saw (due to the double-run), but its definitely possible.

@Stefan Krawczyk note this is not related to the issue you saw — that’s separate and I documented it here: https://github.com/DAGWorks-Inc/hamilton/issues/556