diff --git a/changelog.d/fix-supplied-input-tiers.fixed.md b/changelog.d/fix-supplied-input-tiers.fixed.md new file mode 100644 index 0000000000..5fcbc32d3e --- /dev/null +++ b/changelog.d/fix-supplied-input-tiers.fixed.md @@ -0,0 +1 @@ +- `supplied_input` and `supplied_input_periods` no longer count a value cached over a supplied input as supplied, and read an input stored on disk from disk when a value is cached over it in memory (#2307). diff --git a/policyengine_uk/tests/test_supplied_inputs.py b/policyengine_uk/tests/test_supplied_inputs.py new file mode 100644 index 0000000000..7a1c60e0f6 --- /dev/null +++ b/policyengine_uk/tests/test_supplied_inputs.py @@ -0,0 +1,443 @@ +"""Only a supplied value that is still stored counts as a supplied input. + +``supplied_input`` and ``supplied_input_periods`` tell formulas whether a value +was given to the simulation as an input (``set_input``) rather than filled in +by the engine. policyengine-core keeps its record of an input +(``Simulation._user_input_keys``) after a value is cached over the input, so +the helpers read a supplied value only from a storage tier (memory or disk) +that still holds it (#2307). Before, they counted a value cached over an input +as supplied, and read an input stored on disk from memory when a value was +cached over it there. + +Invariants, for every sequence of inputs set, values cached (calculated or +not), arrays deleted and calculations run, on a branch, on the branch it was +made from and on the default branch, in memory or on disk: + +1. Survival (model): ``supplied_input`` returns the input on the nearest of + those branches that still stores one for the period, reading memory before + disk; otherwise None. An input stays stored until it is deleted, or + another input or a value is cached over it in the same tier. A value + cached in memory over an input on disk leaves the input. A value the + engine calculates never replaces an input. +2. Agreement: ``supplied_input`` returns a value for a period exactly when + ``supplied_input_periods`` lists the period, and the list is sorted + earliest first. +3. Differential: for every recorded input, the helpers count it as supplied + exactly when policyengine-core's ``Holder._stores_user_input`` does. On + policyengine-core 3.33.1 and later, ``supplied_input`` returns what + ``Simulation._get_recorded_input_array`` returns, and from 3.33.2 + ``supplied_input_periods`` lists the periods + ``Simulation._get_set_input_periods`` does. +4. Calculation: calculating, on the default branch and on a branch, leaves + every supplied value of every variable in ``SUPPLIED_INPUT_VARIABLES`` + supplied and unchanged. No calculation caches a value over a supplied + input, so in normal use the helpers gave the right answer before the fix + too. + +Intended: a neutralised variable whose input is still stored reads as its +default, as it does through the engine. +""" + +from contextlib import contextmanager, nullcontext +from types import SimpleNamespace + +import numpy as np +import pytest +from hypothesis import HealthCheck, example, given, settings +from hypothesis import strategies as st +from policyengine_core.data_storage import OnDiskStorage +from policyengine_core.periods import period as period_ +from policyengine_core.reforms import Reform + +from policyengine_uk import Simulation +from policyengine_uk.utils.supplied_inputs import ( + SUPPLIED_INPUT_VARIABLES, + stored_supplied_value, + supplied_input, + supplied_input_periods, +) + +pytestmark = pytest.mark.usefixtures("cloned_uk_tax_benefit_system") + + +def single_person(person=None): + return Simulation( + situation={ + "people": {"person": {"age": {2025: 40}, **(person or {})}}, + "benunits": {"benunit": {"members": ["person"]}}, + "households": {"household": {"members": ["person"]}}, + } + ) + + +@contextmanager +def under_memory_pressure(holder, folder): + """Store what ``holder`` stores next on disk, unless it already holds a + value for that period in memory, as a simulation over its memory limit + does.""" + if holder._disk_storage is None: + holder._disk_storage = OnDiskStorage(str(folder), preserve_storage_dir=True) + holder._on_disk_storable = True + # A memory limit no simulation can stay under. + holder.simulation.memory_config = SimpleNamespace(max_memory_occupation_pc=-1) + try: + yield + finally: + holder._on_disk_storable = False + holder.simulation.memory_config = None + + +def read(population, variable, year): + value = supplied_input(population, variable, period_(year)) + return None if value is None else value.tolist() + + +def recorded_input_array(simulation, variable, year): + """policyengine-core's own reading of the supplied value, from 3.33.1.""" + value = simulation._get_recorded_input_array(variable, period_(year)) + return None if value is None else np.asarray(value).tolist() + + +@pytest.mark.parametrize("variable", sorted(SUPPLIED_INPUT_VARIABLES)) +def test_a_value_cached_over_an_input_is_not_supplied(variable): + sim = single_person() + holder = sim.get_holder(variable) + population = holder.population + sim.set_input(variable, 2025, [100.0]) + assert read(population, variable, 2025) == [100.0] + + holder.put_in_cache(np.array([999.0]), period_(2025)) + + assert holder.get_array(period_(2025)).tolist() == [999.0] + assert read(population, variable, 2025) is None + assert supplied_input_periods(population, variable) == [] + assert not holder._stores_user_input(period_(2025), "default") + if hasattr(sim, "_get_recorded_input_array"): + assert recorded_input_array(sim, variable, 2025) is None + + +def test_other_years_inputs_are_still_supplied(): + sim = single_person({"trading_loss": {2024: 100.0, 2025: 200.0, 2026: 300.0}}) + holder = sim.get_holder("trading_loss") + holder.put_in_cache(np.array([999.0]), period_(2025)) + population = holder.population + assert [read(population, "trading_loss", year) for year in (2024, 2025, 2026)] == [ + [100.0], + None, + [300.0], + ] + assert supplied_input_periods(population, "trading_loss") == [ + period_(2024), + period_(2026), + ] + + +def test_an_input_on_disk_under_a_value_cached_in_memory_is_supplied(tmp_path): + sim = single_person() + holder = sim.get_holder("trading_loss") + with under_memory_pressure(holder, tmp_path): + sim.set_input("trading_loss", 2025, [100.0]) + assert holder._disk_storage.has(period_(2025), "default") + assert not holder._memory_storage.has(period_(2025), "default") + + holder.put_in_cache(np.array([999.0]), period_(2025)) + + # The engine reads the cached value, but the input is still stored. + assert holder.get_array(period_(2025)).tolist() == [999.0] + population = holder.population + assert read(population, "trading_loss", 2025) == [100.0] + assert supplied_input_periods(population, "trading_loss") == [period_(2025)] + assert holder._stores_user_input(period_(2025), "default") + if hasattr(sim, "_get_recorded_input_array"): + assert recorded_input_array(sim, "trading_loss", 2025) == [100.0] + + +def test_a_value_cached_over_an_input_on_disk_is_not_supplied(tmp_path): + sim = single_person() + holder = sim.get_holder("trading_loss") + with under_memory_pressure(holder, tmp_path): + sim.set_input("trading_loss", 2025, [100.0]) + holder.put_in_cache(np.array([999.0]), period_(2025)) + assert holder._disk_storage.get(period_(2025)).tolist() == [999.0] + assert not holder._memory_storage.has(period_(2025), "default") + + population = holder.population + assert read(population, "trading_loss", 2025) is None + assert supplied_input_periods(population, "trading_loss") == [] + assert not holder._stores_user_input(period_(2025), "default") + + +def test_a_calculated_value_is_never_stored_over_an_input(tmp_path): + """policyengine-core stores a calculated (derived) value over nothing the + engine reads as an input, in memory or on disk.""" + sim = single_person() + holder = sim.get_holder("trading_loss") + sim.set_input("trading_loss", 2025, [100.0]) + with under_memory_pressure(holder, tmp_path): + sim.set_input("trading_loss", 2026, [200.0]) + holder.put_in_cache(np.array([999.0]), period_(2026), derived=True) + holder.put_in_cache(np.array([999.0]), period_(2025), derived=True) + holder.put_in_cache(np.array([999.0]), period_(2026), derived=True) + assert holder.get_array(period_(2025)).tolist() == [100.0] + assert holder.get_array(period_(2026)).tolist() == [200.0] + population = holder.population + assert read(population, "trading_loss", 2025) == [100.0] + assert read(population, "trading_loss", 2026) == [200.0] + + +def test_the_nearest_branch_still_storing_an_input_is_read(): + sim = single_person({"trading_loss": {2025: 100.0}}) + branch = sim.get_branch("reform") + population = branch.get_variable_population("trading_loss") + # The branch reads the input of the simulation it was made from. + assert read(population, "trading_loss", 2025) == [100.0] + + branch.set_input("trading_loss", 2025, [200.0]) + assert read(population, "trading_loss", 2025) == [200.0] + # The simulation the branch was made from does not see the branch's input. + assert read(sim.get_variable_population("trading_loss"), "trading_loss", 2025) == [ + 100.0 + ] + + holder = branch.get_holder("trading_loss") + holder.put_in_cache(np.array([999.0]), period_(2025), "reform") + # The branch's input is gone, so the input it was made from is read, as + # policyengine-core reads it. + assert read(population, "trading_loss", 2025) == [100.0] + assert supplied_input_periods(population, "trading_loss") == [period_(2025)] + if hasattr(branch, "_get_recorded_input_array"): + assert recorded_input_array(branch, "trading_loss", 2025) == [100.0] + + holder.put_in_cache(np.array([998.0]), period_(2025), "default") + assert read(population, "trading_loss", 2025) is None + assert supplied_input_periods(population, "trading_loss") == [] + + +class neutralize_trading_loss(Reform): + def apply(self): + self.neutralize_variable("trading_loss") + + +def test_a_neutralised_input_reads_as_its_default_only_while_it_is_stored(): + sim = single_person({"trading_loss": {2025: 100.0, 2026: 200.0}}) + sim.apply_reform(neutralize_trading_loss) + holder = sim.get_holder("trading_loss") + holder.put_in_cache(np.array([999.0]), period_(2026)) + population = holder.population + assert read(population, "trading_loss", 2025) == [0.0] + assert read(population, "trading_loss", 2026) is None + assert supplied_input_periods(population, "trading_loss") == [period_(2025)] + + +# Every variable in SUPPLIED_INPUT_VARIABLES, with a value for 2025 and 2026. +SUPPLIED = { + "bus_fare_spending": 300.0, + "bus_in_london_trips": 40.0, + "other_local_bus_trips": 60.0, + "local_bus_trips": 110.0, + "ni_class_4_losses_brought_forward": 500.0, + "ni_class_4_trading_loss": 200.0, + "trading_loss": 1_000.0, +} +CALCULATED = [ + "household_net_income", + "income_tax", + "ni_class_4", + "ni_class_4_losses_carried_forward", + "ni_class_4_trading_loss", + "ni_class_4_losses_brought_forward", + "trading_loss", + "person_bus_fare_spending", + "bus_fare_spending", + "local_bus_trips", +] + + +def assert_supplied(sim, expected): + for variable in SUPPLIED_INPUT_VARIABLES: + population = sim.get_variable_population(variable) + years = sorted(year for name, year in expected if name == variable) + assert supplied_input_periods(population, variable) == [ + period_(year) for year in years + ] + for year in years: + assert read(population, variable, year) == [expected[variable, year]] + for variable, branch_name, period in sim._user_input_keys: + assert sim.get_holder(variable)._stores_user_input(period, branch_name) + + +def test_calculating_leaves_every_supplied_value_supplied(): + assert set(SUPPLIED) == SUPPLIED_INPUT_VARIABLES + sim = single_person( + { + "self_employment_income": {year: 30_000 for year in range(2024, 2028)}, + "employment_income": {year: 10_000 for year in range(2024, 2028)}, + } + ) + expected = {} + for year in (2025, 2026): + for variable, value in SUPPLIED.items(): + sim.set_input(variable, year, [value + year]) + expected[variable, year] = value + year + branch = sim.get_branch("reform") + branch.set_input("trading_loss", 2027, [123.0]) + branch_expected = {**expected, ("trading_loss", 2027): 123.0} + + for simulation in (sim, branch): + for year in (2024, 2025, 2026, 2027, 2028): + for variable in CALCULATED: + simulation.calculate(variable, year) + + assert_supplied(sim, expected) + assert_supplied(branch, branch_expected) + + +BRANCHES = ("child", "parent", "default") # nearest first +YEARS = (2024, 2025, 2026) +step = st.one_of( + # Set an input, on disk if under memory pressure. + st.tuples( + st.just("set"), st.sampled_from(BRANCHES), st.sampled_from(YEARS), st.booleans() + ), + # Cache a value, calculated (derived) or not, on disk if under pressure. + st.tuples( + st.just("cache"), + st.sampled_from(BRANCHES), + st.sampled_from(YEARS), + st.booleans(), + st.booleans(), + ), + st.tuples(st.just("delete"), st.sampled_from(BRANCHES), st.sampled_from(YEARS)), + st.tuples( + st.just("calculate"), + st.sampled_from(["trading_loss", "ni_class_4_losses_brought_forward"]), + st.sampled_from(YEARS + (2027,)), + ), +) + +_templates = [] + + +def branch_chain(): + """A branch ("child") of a branch ("parent") of a simulation.""" + if not _templates: + _templates.append(single_person()) + root = _templates[0].clone(clone_tax_benefit_system=False) + return root.get_branch("parent").get_branch("child") + + +def tier_holding(holder, period, branch_name, value): + """The tier that holds ``value`` for ``period`` on ``branch_name``.""" + for tier, storage in ( + ("memory", holder._memory_storage), + ("disk", holder._disk_storage), + ): + if ( + storage is not None + and storage.has(period, branch_name) + and storage.get(period, branch_name).tolist() == [value] + ): + return tier + return None + + +def expected_supplied(stored, year): + """Invariant 1: the input on the nearest branch, memory before disk.""" + for branch_name in BRANCHES: + for tier in ("memory", "disk"): + if (branch_name, year, tier) in stored: + return [stored[branch_name, year, tier]] + return None + + +@settings( + max_examples=100, + deadline=None, + derandomize=True, + suppress_health_check=[HealthCheck.too_slow], +) +@given(steps=st.lists(step, max_size=10)) +# A value cached over the input in the same tier (#2307). +@example(steps=[("set", "child", 2025, False), ("cache", "child", 2025, False, False)]) +# An input on disk under a value cached in memory. +@example(steps=[("set", "child", 2025, True), ("cache", "child", 2025, False, False)]) +# A value cached over the child's input leaves the parent's. +@example( + steps=[ + ("set", "parent", 2025, False), + ("set", "child", 2025, True), + ("cache", "child", 2025, False, True), + ] +) +# An input in memory over an earlier input on disk is read, as the engine +# reads it. +@example(steps=[("set", "child", 2025, True), ("set", "child", 2025, False)]) +# From policyengine-core 3.33, the first input drops the cached value, so it +# and the second input go to disk (found by Hypothesis). +@example( + steps=[ + ("cache", "child", 2024, False, False), + ("set", "child", 2024, True), + ("set", "child", 2024, True), + ] +) +def test_only_inputs_still_stored_are_supplied(tmp_path_factory, steps): + child = branch_chain() + holder = child.get_holder("trading_loss") + population = holder.population + folder = tmp_path_factory.mktemp("disk") + # The surviving inputs, by (branch, year, tier). + stored = {} + for index, (kind, *args) in enumerate(steps): + if kind in ("set", "cache"): + branch_name, year, *flags = args + pressure = flags[-1] + period = period_(year) + # Every value written is new, so the tier holding it shows where + # core stored it. (Under pressure, core stores on disk unless + # memory holds a value for the period; from 3.33, setting an input + # first drops cached values that may depend on it.) + value = (1.0 if kind == "set" else 10_000.0) + index + with under_memory_pressure(holder, folder) if pressure else nullcontext(): + if kind == "set": + holder.set_input(period, np.array([value]), branch_name) + else: + derived = flags[0] + holder.put_in_cache(np.array([value]), period, branch_name, derived) + tier = tier_holding(holder, period, branch_name, value) + if kind == "set": + assert tier is not None + stored[branch_name, year, tier] = value + elif tier is not None: + # A calculated value never replaces an input. + assert not (derived and (branch_name, year, tier) in stored) + stored.pop((branch_name, year, tier), None) + elif kind == "delete": + branch_name, year = args + holder.delete_arrays(period_(year), branch_name) + for tier in ("memory", "disk"): + stored.pop((branch_name, year, tier), None) + else: + variable, year = args + child.calculate(variable, year) + + supplied_years = [] + for year in YEARS: + expected = expected_supplied(stored, year) + assert read(population, "trading_loss", year) == expected, (year, stored) + if expected is not None: + supplied_years.append(year) + if hasattr(child, "_get_recorded_input_array"): + assert recorded_input_array(child, "trading_loss", year) == expected + # Invariant 2. + periods = supplied_input_periods(population, "trading_loss") + assert periods == [period_(year) for year in supplied_years] + if hasattr(child, "_get_set_input_periods"): + assert set(child._get_set_input_periods("trading_loss")) == set(periods) + # Invariant 3, for every recorded input of the variable. + for variable, branch_name, period in child._user_input_keys: + if variable == "trading_loss": + ours = stored_supplied_value(holder, period, branch_name) + assert (ours is not None) == holder._stores_user_input( + period, branch_name + ), (branch_name, period) diff --git a/policyengine_uk/utils/supplied_inputs.py b/policyengine_uk/utils/supplied_inputs.py index e9b7a5cdcf..dadba245b9 100644 --- a/policyengine_uk/utils/supplied_inputs.py +++ b/policyengine_uk/utils/supplied_inputs.py @@ -7,8 +7,18 @@ (variable, branch, period) that ``set_input`` stores in ``Simulation._user_input_keys``. From 3.32.27 (policyengine-core#561), core forgets the record of an input that ``Holder.delete_arrays`` deletes, and each -clone gets its own copy of the record. The helpers also require a stored -value, so missing arrays do not count as supplied inputs. +clone gets its own copy of the record. + +The record can still outlive the value it names. A value cached over an input +takes that storage tier (memory or disk) off the holder's record of where the +input is (``Holder._user_input_storage``), but leaves the entry in +``_user_input_keys``. So every helper here reads a supplied value only if it is +still stored, and only from a tier that the holder records the input in and +that holds a value the engine did not calculate, as +``Holder._stores_user_input`` decides whether an input survives (see +``stored_supplied_value``). A missing value, or one carried over, calculated or +cached into the same tier, is not supplied; an input stored on disk under a +value cached in memory still is. ``build_from_situation`` installs fresh populations, and so fresh holders, through ``Simulation.build_from_populations``, which keeps the record @@ -20,10 +30,7 @@ ``Simulation.delete_arrays`` forgets such records too, for any variable it deletes. -``supplied_input_arrays`` answers for every variable. It reads only a record -whose supplied value is still stored, from the storage tier (memory or disk) -the holder records it in. A value the engine carried over or calculated into -the same period, in either tier, is not read as supplied. +``supplied_input_arrays`` answers for every variable. """ from typing import List, Optional, Tuple @@ -88,7 +95,9 @@ def drop_missing_supplied_inputs(simulation, variable_name: str) -> None: def supplied_input_periods(population, variable_name: str) -> List[Period]: """Periods for which ``variable_name`` was supplied as an input, on this - branch or one it was made from, earliest first.""" + branch or one it was made from, and the supplied value is still stored, + earliest first. A period whose input a value carried over, calculated or + cached in its place has replaced is left out.""" _check_registered(variable_name) simulation = population.simulation branch_names = set(_visible_branch_names(simulation)) @@ -100,7 +109,8 @@ def supplied_input_periods(population, variable_name: str) -> List[Period]: for name, branch_name, period in input_keys if name == variable_name and branch_name in branch_names - and holder._get_array_from_storage(period, branch_name) is not None + # Checked without reading the value, which may be on disk. + and _supplied_value_storage(holder, period, branch_name) is not None }, key=lambda period: period.start, ) @@ -110,9 +120,10 @@ def supplied_input( population, variable_name: str, period: Period ) -> Optional[np.ndarray]: """The value of ``variable_name`` for ``period`` if it was supplied as an - input, on this branch or the nearest one it was made from; otherwise - None. A value the engine carried over from an earlier period, or cached - after a calculation, is not an input.""" + input, on this branch or the nearest one it was made from, and is still + stored; otherwise None. A value the engine carried over from an earlier + period, or cached after a calculation, is not an input, including one + cached over the input.""" _check_registered(variable_name) simulation = population.simulation input_keys = getattr(simulation, "_user_input_keys", None) or () @@ -120,10 +131,12 @@ def supplied_input( for branch_name in _visible_branch_names(simulation): if (variable_name, branch_name, period) in input_keys: # Read the stored input itself: Holder.get_array would fall back - # to other branches' values, including cached calculations. - # A record can outlive its stored array (see above); a missing - # stored array must be ignored even if its key remains. - value = holder._get_array_from_storage(period, branch_name) + # to other branches' values, including cached calculations, and + # reads memory before disk whatever each holds. A record can + # outlive its input (see above); then look on the branch this one + # was made from, as policyengine-core's + # Simulation._get_recorded_input_array does. + value = stored_supplied_value(holder, period, branch_name) if value is not None: # A neutralized variable reads as its default, as it does # through the engine. (PolicyEngine-UK builds no @@ -167,6 +180,13 @@ def stored_supplied_value(holder, period: Period, branch_name: str): Reading memory before disk instead could return a calculated value cached over an input stored on disk. """ + storage = _supplied_value_storage(holder, period, branch_name) + return None if storage is None else storage.get(period, branch_name) + + +def _supplied_value_storage(holder, period: Period, branch_name: str): + """The storage (memory or disk) from which ``stored_supplied_value`` + reads, or None if the supplied value is gone.""" recorded = (getattr(holder, "_user_input_storage", None) or {}).get( (branch_name, str(period)) ) @@ -181,5 +201,5 @@ def stored_supplied_value(holder, period: Period, branch_name: str): and storage.has(period, branch_name) and not storage.is_derived(period, branch_name) ): - return storage.get(period, branch_name) + return storage return None