Use explicit pydantic model in selector business logic (#454)

Sorry for this bulky PR; this is really not my style :(

RHO-LOSS needs more sophisticated parameters to the downsampling config.
We currently use `dict` in business code to store the config, which is
hard to reason, error-prune, and too cumbersome to further support
RHO-LOSS.

Therefore, In this PR I changed the occurrence of all `dict` to the
appropriate pydantic models, and also simplified some selection config
settings.

After this PR, we will almost have replaced every occurrence of using
`dict` as the config with the pydantic model!

benchmark/mnist/mnist.yaml

@@ -34,7 +34,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
 data:
   dataset_id: mnist
   transformations: ["transforms.ToTensor()",

benchmark/wildtime_benchmarks/example_pipelines/arxiv.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 10000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
 data:
   dataset_id: arxiv
   bytes_parser_function: |

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/arxiv_datadrift.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 10000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
   seed: 42
   epochs_per_trigger: 1
 data:

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/huffpost_datadrift.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
   seed: 42
   epochs_per_trigger: 1
 data:

benchmark/wildtime_benchmarks/example_pipelines/data_drift_trigger/yearbook_datadrift.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
   seed: 42
   epochs_per_trigger: 1
 data:

benchmark/wildtime_benchmarks/example_pipelines/fmow.yaml

@@ -34,7 +34,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
 data:
   dataset_id: fmow
   transformations: [

benchmark/wildtime_benchmarks/example_pipelines/huffpost.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
 data:
   dataset_id: huffpost
   bytes_parser_function: |

benchmark/wildtime_benchmarks/example_pipelines/yearbook.yaml

@@ -35,7 +35,7 @@ training:
     maximum_keys_in_memory: 1000
     storage_backend: "database"
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
 data:
   dataset_id: yearbook
   transformations: []

experiments/yearbook/compare_trigger_policies/pipeline_config.py

@@ -8,7 +8,7 @@
     Metric,
     ModelConfig,
     ModynPipelineConfig,
-    NewDataSelectionStrategy,
+    NewDataStrategyConfig,
     OptimizationCriterion,
     OptimizerConfig,
     OptimizerParamGroup,
@@ -41,8 +41,8 @@ def gen_pipeline_config(name: str, trigger: TriggerConfig, eval_strategy: EvalSt
             ],
             optimization_criterion=OptimizationCriterion(name="CrossEntropyLoss"),
             checkpointing=CheckpointingConfig(activated=False),
-            selection_strategy=NewDataSelectionStrategy(
-                maximum_keys_in_memory=1000, storage_backend="database", limit=-1, reset_after_trigger=True
+            selection_strategy=NewDataStrategyConfig(
+                maximum_keys_in_memory=1000, storage_backend="database", limit=-1, tail_triggers=0
             ),
         ),
         data=DataConfig(

integrationtests/online_dataset/test_online_dataset.py

@@ -19,7 +19,7 @@
     init_metadata_db,
     register_pipeline,
 )
-from modyn.config.schema.pipeline import NewDataSelectionStrategy
+from modyn.config.schema.pipeline import NewDataStrategyConfig
 from modyn.selector.internal.grpc.generated.selector_pb2 import DataInformRequest
 from modyn.selector.internal.grpc.generated.selector_pb2_grpc import SelectorStub
 from modyn.storage.internal.grpc.generated.storage_pb2 import (
@@ -205,8 +205,8 @@ def prepare_selector(num_dataworkers: int, keys: list[int]) -> Tuple[int, int]:
     # We test the NewData strategy for finetuning on the new data, i.e., we reset without limit
     # We also enforce high partitioning (maximum_keys_in_memory == 2) to ensure that works
 
-    strategy_config = NewDataSelectionStrategy(
-        maximum_keys_in_memory=2, limit=-1, reset_after_trigger=True, storage_backend="database"
+    strategy_config = NewDataStrategyConfig(
+        maximum_keys_in_memory=2, limit=-1, tail_triggers=0, storage_backend="database"
     )
     pipeline_config = get_minimal_pipeline_config(max(num_dataworkers, 1), strategy_config.model_dump(by_alias=True))
     init_metadata_db(get_modyn_config())

integrationtests/selector/integrationtest_selector.py

@@ -1,11 +1,11 @@
 import grpc
 from integrationtests.utils import get_minimal_pipeline_config, get_modyn_config, init_metadata_db, register_pipeline
 from modyn.config.schema.pipeline import (
-    CoresetSelectionStrategy,
-    DownsamplingConfig,
-    NewDataSelectionStrategy,
+    CoresetStrategyConfig,
+    NewDataStrategyConfig,
     PresamplingConfig,
 )
+from modyn.config.schema.sampling.downsampling_config import LossDownsamplingConfig
 from modyn.selector.internal.grpc.generated.selector_pb2 import (
     DataInformRequest,
     GetAvailableLabelsRequest,
@@ -19,21 +19,21 @@
 # TODO(54): Write more integration tests for different strategies.
 
 
-def get_coreset_strategy_config() -> CoresetSelectionStrategy:
-    return CoresetSelectionStrategy(
+def get_coreset_strategy_config() -> CoresetStrategyConfig:
+    return CoresetStrategyConfig(
         maximum_keys_in_memory=250,
         storage_backend="database",  # TODO(#324): Support local backend
         limit=-1,
-        reset_after_trigger=True,
+        tail_triggers=0,
         presampling_config=PresamplingConfig(strategy="LabelBalanced", ratio=50),
     )
 
 
-def get_newdata_strategy_config() -> NewDataSelectionStrategy:
-    return NewDataSelectionStrategy(
+def get_newdata_strategy_config() -> NewDataStrategyConfig:
+    return NewDataStrategyConfig(
         maximum_keys_in_memory=2,
         limit=-1,
-        reset_after_trigger=True,
+        tail_triggers=0,
         storage_backend="database",
     )
 
@@ -122,7 +122,7 @@ def test_label_balanced_force_same_size():
 
     strategy_config = get_coreset_strategy_config()
     strategy_config.maximum_keys_in_memory = 100
-    strategy_config.reset_after_trigger = False
+    strategy_config.tail_triggers = None
     strategy_config.presampling_config.force_column_balancing = True
     strategy_config.presampling_config.ratio = 90
 
@@ -197,7 +197,7 @@ def test_label_balanced_force_all_samples():
 
     strategy_config = get_coreset_strategy_config()
     strategy_config.maximum_keys_in_memory = 100
-    strategy_config.reset_after_trigger = False
+    strategy_config.tail_triggers = None
     strategy_config.presampling_config.force_required_target_size = True
     strategy_config.presampling_config.ratio = 90
 
@@ -405,17 +405,17 @@ def test_newdata() -> None:
     assert len(total_samples) == 6
 
 
-def test_abstract_downsampler(reset_after_trigger) -> None:
+def test_abstract_downsampler(reset_after_trigger: bool) -> None:
     selector_channel = connect_to_selector_servicer()
     selector = SelectorStub(selector_channel)
 
     # sampling every datapoint
     strategy_config = get_coreset_strategy_config()
     strategy_config.maximum_keys_in_memory = 50000
-    strategy_config.reset_after_trigger = reset_after_trigger
+    strategy_config.tail_triggers = 0 if reset_after_trigger else None
     strategy_config.presampling_config.ratio = 20
     strategy_config.presampling_config.strategy = "Random"
-    strategy_config.downsampling_config = DownsamplingConfig(strategy="Loss", ratio=10, sample_then_batch=False)
+    strategy_config.downsampling_config = LossDownsamplingConfig(ratio=10, sample_then_batch=False)
 
     pipeline_config = get_minimal_pipeline_config(2, strategy_config.model_dump(by_alias=True))
     pipeline_id = register_pipeline(pipeline_config, get_modyn_config())
@@ -561,7 +561,7 @@ def test_empty_triggers() -> None:
 
     # TODO(MaxiBoether): use local strategy here as well after implementing it
     strategy_config = get_newdata_strategy_config()
-    strategy_config.reset_after_trigger = False
+    strategy_config.tail_triggers = None
 
     pipeline_config = get_minimal_pipeline_config(2, strategy_config.model_dump(by_alias=True))
     pipeline_id = register_pipeline(pipeline_config, get_modyn_config())
@@ -727,7 +727,7 @@ def test_many_samples_evenly_distributed():
     # TODO(MaxiBoether): use local strategy here as well after implementing it
     strategy_config = get_newdata_strategy_config()
     strategy_config.maximum_keys_in_memory = 5000
-    strategy_config.reset_after_trigger = False
+    strategy_config.tail_triggers = None
 
     pipeline_config = get_minimal_pipeline_config(2, strategy_config.model_dump(by_alias=True))
     pipeline_id = register_pipeline(pipeline_config, get_modyn_config())
@@ -795,7 +795,7 @@ def test_many_samples_unevenly_distributed():
     # TODO(MaxiBoether): use local strategy here as well after implementing it
     strategy_config = get_newdata_strategy_config()
     strategy_config.maximum_keys_in_memory = 4999
-    strategy_config.reset_after_trigger = False
+    strategy_config.tail_triggers = None
 
     pipeline_config = get_minimal_pipeline_config(2, strategy_config.model_dump(by_alias=True))
     pipeline_id = register_pipeline(pipeline_config, get_modyn_config())
@@ -864,7 +864,7 @@ def test_get_available_labels(reset_after_trigger: bool):
     # TODO(MaxiBoether): use local strategy here as well after implementing it
     strategy_config = get_newdata_strategy_config()
     strategy_config.maximum_keys_in_memory = 2
-    strategy_config.reset_after_trigger = reset_after_trigger
+    strategy_config.tail_triggers = 0 if reset_after_trigger else None
 
     pipeline_config = get_minimal_pipeline_config(2, strategy_config.model_dump(by_alias=True))
     pipeline_id = register_pipeline(pipeline_config, get_modyn_config())

modyn/config/__init__.py

@@ -1,6 +1,10 @@
 from pathlib import Path
 
 import yaml
+from modyn.config.schema.sampling.downsampling_config import (  # noqa: F401
+    MultiDownsamplingConfig,
+    SingleDownsamplingConfig,
+)
 
 from .schema.config import (
     BinaryFileByteOrder,
@@ -24,19 +28,20 @@
     TensorboardConfig,
     TrainingServerConfig,
 )
-from .schema.pipeline import (
+from .schema.pipeline import (  # noqa: F401
     CheckpointingConfig,
+    CoresetStrategyConfig,
     DataConfig,
-    DownsamplingConfig,
     EvalDataConfig,
     EvaluationConfig,
+    FreshnessSamplingStrategyConfig,
     FullModelStrategy,
     IncrementalModelStrategy,
     LrSchedulerConfig,
     Metric,
     ModelConfig,
     ModynPipelineConfig,
-    MultiDownsamplingConfig,
+    NewDataStrategyConfig,
     OptimizationCriterion,
     OptimizerConfig,
     OptimizerParamGroup,
@@ -79,8 +84,8 @@
     "IncrementalModelStrategy",
     "PipelineModelStorageConfig",
     "PresamplingConfig",
-    "DownsamplingConfig",
-    "MultiDownsamplingConfig",
+    "SingleDownsamplingConfig",
+    "FreshnessSamplingStrategyConfig",
     "SelectionStrategy",
     "CheckpointingConfig",
     "OptimizerParamGroup",

modyn/config/examples/example-pipeline.yaml

@@ -50,7 +50,7 @@ training:
     name: NewDataStrategy
     maximum_keys_in_memory: 500000
     limit: -1
-    reset_after_trigger: True
+    tail_triggers: 0
     storage_backend: database
     processor_type: basic_processor_strategy
 data:

modyn/config/schema/modyn_base_model.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class ModynBaseModel(BaseModel):
+    class Config:
+        extra = "forbid"