Feature/custom solver support

docs/impulse/docs/data_model/ingestion.md

@@ -233,7 +233,8 @@ computed-column joins, JSON-encoded values, multi-column composite keys
 that need pre-processing, etc. — you can implement a custom solver by
 subclassing
 [`QuerySolver`](../references/api/impulse_query_engine/analyze/query/solvers/query_solver.md)
-(or one of the existing solvers) and registering it in your report config.
+(or one of the existing solvers, usually `DefaultSolver`) and selecting it
+by name in your report config.
 
 This is significantly more invested than the `SolverConfig` path: you take
 on responsibility for the four solver pipeline stages
@@ -245,6 +246,110 @@ yourself heading down this path, it is usually worth first asking whether
 a one-time ETL job to produce the standard silver-layer shape would be
 cheaper.
 
+#### Registering a custom solver
+
+Decorate your subclass with `@register_solver(name)`. The report config then
+selects it by that name via the existing `query_engine.solver` field — the
+same field used for the built-in `DefaultSolver`.
+
+```python
+from impulse_query_engine.analyze.query.solvers import (
+    register_solver,
+    SolverConfig,
+)
+from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver
+
+
+class CustomSolverConfig(SolverConfig):
+    raw_signal_table: str            # required — validated when the config is parsed
+    position_signal_name: str = "POSITION"  # optional, with a default
+
+
+@register_solver("CustomSolver", CustomSolverConfig)
+class CustomSolver(DefaultSolver):
+    """Reshapes a custom raw table into the Impulse silver schema."""
+    ...
+```
+
+```json
+{
+  "query_engine": {
+    "solver": "CustomSolver",
+    "solver_config": {
+      "raw_signal_table": "my_catalog.my_schema.raw_signals",
+      "channels": { "column_name_mapping": { "signal_value": "value" } }
+    }
+  }
+}
+```
+
+Passing `CustomSolverConfig` to `register_solver` makes Impulse validate the
+`solver_config` block through that subclass, so its extra fields are checked
+at config-load time: a missing required `raw_signal_table`, or a mistyped
+key, raises a `ValidationError` up front rather than being silently dropped.
+Inside the solver you read them with normal typed attribute access
+(`self.config.raw_signal_table`).
+
+**Loading the solver.** `@register_solver` runs only when its module is
+imported, so import your solver package once in the driver before building
+the report — that import is what registers the name:
+
+```python
+import sys
+sys.path.append("/Workspace/Repos/me/my_solvers_root")  # dir containing my_solvers/
+
+import my_solvers.custom_solver       # runs @register_solver("CustomSolver")
+
+from impulse_reporting.core.report import Report
+
+report = Report(name="my_report", spark=spark, workspace_client=ws,
+                config_path="report_config.json")
+report.determine_report()
+```
+
+If the name is not registered when the config is parsed, Impulse raises an
+error listing the known names — a quick signal that the import is missing or
+misspelled. (For an installed wheel the `sys.path` line is unnecessary; the
+`import` alone suffices. In a notebook you can also just define the solver in
+a cell — running that cell registers it.)
+
+Selection is by registered name only: a report config can never, on its own,
+cause Impulse to import an arbitrary class. Which solvers exist is governed
+entirely by what the driver imports.
+
+#### Example: a column-redaction solver
+
+A solver override plus one config field gives a config-switched masking layer:
+
+```python
+import pyspark.sql.functions as F
+from impulse_query_engine.analyze.query.solvers import register_solver, SolverConfig
+from impulse_query_engine.analyze.query.solvers.default_solver import DefaultSolver
+
+
+class RedactConfig(SolverConfig):
+    redact_columns: list[str] = []
+
+
+@register_solver("RedactingSolver", RedactConfig)
+class RedactingSolver(DefaultSolver):
+    def solve(self, query, channels_df, selections, dtypes):
+        df = super().solve(query, channels_df, selections, dtypes)
+        for col in self.config.redact_columns:
+            if col in df.columns:
+                df = df.withColumn(col, F.lit(None).cast(df.schema[col].dataType))
+        return df
+```
+
+```json
+{
+  "query_engine": {
+    "solver": "RedactingSolver",
+    "solver_config": { "redact_columns": ["vin", "gps_lat", "gps_lon"] }
+  }
+}
+```
+
 The general rule: **`SolverConfig` for naming differences, custom solver
 for structural differences, ETL into the standard shape for everything
 else.**

src/impulse_query_engine/analyze/query/solvers/__init__.py

@@ -1,4 +1,24 @@
 from .default_solver import DefaultSolver, TimeSeriesCache
+from .query_solver import QuerySolver
+from .registry import (
+    SolverRegistration,
+    is_registered,
+    register_solver,
+    registered_names,
+    resolve_registration,
+)
 from .solver_config import SolverConfig
+from .solver_context import SolverBuildContext
 
-__all__ = ["DefaultSolver", "TimeSeriesCache", "SolverConfig"]
+__all__ = [
+    "DefaultSolver",
+    "TimeSeriesCache",
+    "SolverConfig",
+    "QuerySolver",
+    "SolverBuildContext",
+    "SolverRegistration",
+    "register_solver",
+    "resolve_registration",
+    "is_registered",
+    "registered_names",
+]

src/impulse_query_engine/analyze/query/solvers/default_solver.py

@@ -14,6 +14,7 @@
 from impulse_query_engine.model.series.sample_series import SampleSeries
 
 from .query_solver import QuerySolver
+from .registry import register_solver
 from .series_cache import SeriesCache
 from .solver_config import RawEncoder, SolverConfig
 from .utils.interval_encoder import IntervalEncoder
@@ -22,6 +23,8 @@
 if TYPE_CHECKING:
     from impulse_query_engine.measurement_db import MeasurementDB
 
+    from .solver_context import SolverBuildContext
+
 
 class TimeSeriesCache(SeriesCache):
     def __init__(self, pdf, col_map: dict[str, str]):
@@ -134,6 +137,7 @@ def load_blob(self, mid, cid, uses_alias: bool = False):
         return SampleSeries(s[self._ts_col], s[self._te_col], values)
 
 
+@register_solver("DefaultSolver", aliases=("DeltaSolver", "KeyValueStoreSolver"))
 class DefaultSolver(QuerySolver):
     """
     The default query-engine solver.  Adapts to the shape of the silver layer.
@@ -197,6 +201,21 @@ def __init__(
         self.raw_encoder: RawEncoder = raw_encoder
         self.channel_encoder: RleEncoder | IntervalEncoder = self._build_channel_encoder()
 
+    @classmethod
+    def from_config(cls, ctx: "SolverBuildContext") -> "DefaultSolver":
+        """Build a :class:`DefaultSolver` from a :class:`SolverBuildContext`.
+
+        Overrides the base hook to wire the SparkSession and the raw-data flags
+        that this solver requires beyond ``solver_config``.
+        """
+        return cls(
+            ctx.spark,
+            config=ctx.solver_config,
+            is_raw_data=ctx.is_raw_data,
+            drop_implausible_data=ctx.drop_implausible_data,
+            raw_encoder=ctx.raw_encoder,
+        )
+
     def _build_channel_encoder(self) -> RleEncoder | IntervalEncoder:
         """Construct the raw -> interval encoder selected by ``raw_encoder``.
 
@@ -261,7 +280,7 @@ def filter_container_tags(self, spark, query) -> DataFrame:
                 required_elements.extend(filt.required_tags())
         required_elements = set(required_elements)
 
-        tags = query.db.container_tags(self.spark)
+        tags = self.load_container_tags(query.db, self.spark)
         tags = self._apply_column_mapping(tags, self.config.container_tags.column_name_mapping)
 
         if self.config.project_id is not None:
@@ -329,7 +348,7 @@ def filter_container_metrics(
         if pre_filtered_containers_df is not None:
             metrics = pre_filtered_containers_df
         else:
-            metrics = query.db.container_metrics(self.spark)
+            metrics = self.load_container_metrics(query.db, self.spark)
 
         metrics = self._apply_column_mapping(
             metrics, self.config.container_metrics.column_name_mapping
@@ -403,7 +422,7 @@ def filter_channel_tags(self, spark, db, container_df, selectors) -> DataFrame:
         for selector in selectors:
             required_tags.update(selector.required_tags())
 
-        tbl = db.channel_tags(spark)
+        tbl = self.load_channel_tags(db, spark)
         tbl = self._apply_column_mapping(tbl, self.config.channel_tags.column_name_mapping)
         expr = self._build_expr(selectors)
 
@@ -456,7 +475,7 @@ def filter_channel_metrics(self, spark, db, channel_df, selectors) -> DataFrame:
         if len(selectors) == 0:
             return self._empty_channel_match_df(spark, db)
 
-        channel_metrics_df = db.channel_metrics(spark)
+        channel_metrics_df = self.load_channel_metrics(db, spark)
         channel_metrics_df = self._apply_column_mapping(
             channel_metrics_df, self.config.channel_metrics.column_name_mapping
         )
@@ -537,7 +556,7 @@ def filter_aliased_channel_metrics(
         if len(selectors) == 0:
             return self._empty_channel_match_df(spark, db)
 
-        channel_mapping = db.channel_mapping(spark)
+        channel_mapping = self.load_channel_mapping(db, spark)
         channel_mapping = self._apply_column_mapping(
             channel_mapping, self.config.channel_mapping.column_name_mapping
         )
@@ -552,7 +571,7 @@ def filter_aliased_channel_metrics(
 
         resolved_mapping = channel_mapping.where(self._build_expr(selectors))
 
-        channel_metrics = db.channel_metrics(spark)
+        channel_metrics = self.load_channel_metrics(db, spark)
         channel_metrics = self._apply_column_mapping(
             channel_metrics, self.config.channel_metrics.column_name_mapping
         )
@@ -850,7 +869,7 @@ def _compute_conversion_factors(self, spark, query, channels_df: DataFrame) -> D
             :meth:`_validate_unit_conversion_table` for the underlying
             check.
         """
-        uc_table = query.db.unit_conversion(spark)
+        uc_table = self.load_unit_conversion(query.db, spark)
         uc_table = self._apply_column_mapping(
             uc_table, self.config.unit_conversion.column_name_mapping
         )
@@ -1008,7 +1027,7 @@ def _prepare_channels_join(self, query, channels_df) -> tuple[DataFrame, DataFra
             if col_name in channels_df.columns:
                 channels_df = channels_df.drop(col_name)
 
-        q = query.db.channels(self.spark)
+        q = self.load_channels(query.db, self.spark)
         q = self._apply_column_mapping(q, self.config.channels.column_name_mapping)
 
         if self.is_raw_data:

src/impulse_query_engine/analyze/query/solvers/query_solver.py

@@ -12,6 +12,8 @@
 if TYPE_CHECKING:
     from impulse_query_engine.measurement_db import MeasurementDB
 
+    from .solver_context import SolverBuildContext
+
 from impulse_query_engine.analyze.metadata.time_series_expression import (
     TimeSeriesSelector,
 )
@@ -35,6 +37,57 @@ class QuerySolver(ABC):
     def __init__(self, config: SolverConfig = None):
         self.config = config or SolverConfig()
 
+    @classmethod
+    def from_config(cls, ctx: "SolverBuildContext") -> "QuerySolver":
+        """Build a solver from a :class:`SolverBuildContext`.
+
+        This is the uniform instantiation hook used by the report factory so
+        that solvers with different constructor signatures can all be built the
+        same way.  The base implementation passes only ``solver_config`` (the
+        common case for a :class:`QuerySolver` subclass); solvers that need more
+        construction inputs — such as :class:`DefaultSolver`, which needs the
+        SparkSession and the raw-data flags — override this classmethod.
+
+        Parameters
+        ----------
+        ctx : SolverBuildContext
+            The construction inputs resolved from the report configuration.
+
+        Returns
+        -------
+        QuerySolver
+            A ready-to-use solver instance.
+        """
+        return cls(config=ctx.solver_config)
+
+    def load_container_tags(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``container_tags`` (narrow/EAV) table."""
+        return db.container_tags(spark)
+
+    def load_container_metrics(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``container_metrics`` table."""
+        return db.container_metrics(spark)
+
+    def load_channel_tags(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``channel_tags`` (narrow/EAV) table."""
+        return db.channel_tags(spark)
+
+    def load_channel_metrics(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``channel_metrics`` table."""
+        return db.channel_metrics(spark)
+
+    def load_channel_mapping(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``channel_mapping`` (alias) table."""
+        return db.channel_mapping(spark)
+
+    def load_unit_conversion(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``unit_conversion`` table."""
+        return db.unit_conversion(spark)
+
+    def load_channels(self, db: "MeasurementDB", spark) -> DataFrame:
+        """Read the raw ``channels`` (time-series data) table."""
+        return db.channels(spark)
+
     @staticmethod
     def _apply_column_mapping(df: DataFrame, mapping: dict[str, str]) -> DataFrame:
         """Rename DataFrame columns according to a physical → internal mapping."""
@@ -155,7 +208,7 @@ def _empty_channel_match_df(self, spark, db: MeasurementDB) -> DataFrame:
         schemas stay compatible regardless of the physical id types.
         """
         ref = self._apply_column_mapping(
-            db.channel_metrics(spark), self.config.channel_metrics.column_name_mapping
+            self.load_channel_metrics(db, spark), self.config.channel_metrics.column_name_mapping
         )
         return spark.createDataFrame(
             [],