Static filter (#373)

Static_filters support

---------

Signed-off-by: s0nicboOm <[email protected]>

numalogic/connectors/_config.py

@@ -36,9 +36,16 @@ class Pivot:
     value: list[str] = field(default_factory=lambda: ["count"])
 
 
+@dataclass
+class FilterConf:
+    inclusion_filters: Optional[list[dict]] = None
+    exclusion_filters: Optional[list[dict]] = None
+
+
 @dataclass
 class DruidFetcherConf:
     datasource: str
+    static_filters: Optional[FilterConf] = None
     dimensions: list[str] = field(default_factory=list)
     aggregations: dict = field(default_factory=dict)
     group_by: list[str] = field(default_factory=list)

numalogic/connectors/druid/_druid.py

@@ -10,7 +10,7 @@
 from pydruid.utils.filters import Filter
 
 from numalogic.connectors._base import DataFetcher
-from numalogic.connectors._config import Pivot
+from numalogic.connectors._config import Pivot, FilterConf
 from typing import Optional, Final
 
 from numalogic.tools.exceptions import DruidFetcherError
@@ -34,13 +34,30 @@ def make_filter_pairs(filter_keys: list[str], filter_values: list[str]) -> dict[
     return dict(zip(filter_keys, filter_values))
 
 
+def _combine_in_filters(filters_list) -> Filter:
+    return Filter(type="and", fields=[Filter(**item) for item in filters_list])
+
+
+def _combine_ex_filters(filters_list) -> Filter:
+    filters = _combine_in_filters(filters_list)
+    return Filter(type="not", field=filters)
+
+
+def _make_static_filters(filters: FilterConf) -> Filter:
+    in_filters, ex_filters = _combine_in_filters(filters.inclusion_filters), _combine_ex_filters(
+        filters.exclusion_filters
+    )
+    return Filter(type="and", fields=[in_filters, ex_filters])
+
+
 def build_params(
     datasource: str,
     dimensions: list[str],
     filter_pairs: dict,
     granularity: str,
     hours: float,
     delay: float,
+    static_filters: Optional[FilterConf] = None,
     aggregations: Optional[list[str]] = None,
     post_aggregations: Optional[list[str]] = None,
     reference_dt: Optional[datetime] = None,
@@ -52,6 +69,7 @@ def build_params(
         dimensions: The dimensions to group by
         filter_pairs: Indicates which rows of
           data to include in the query
+        static_filters: Static filters passed from config
         granularity: Time bucket to aggregate data by hour, day, minute, etc.,
         hours: Hours from now to skip training.
         delay: Added delay to the fetch query from current time.
@@ -69,6 +87,11 @@ def build_params(
         type="and",
         fields=[Filter(type="selector", dimension=k, value=v) for k, v in filter_pairs.items()],
     )
+    if static_filters:
+        _LOGGER.debug("Static Filters are present!")
+        _static_filters = _make_static_filters(static_filters)
+        _filter = Filter(type="and", fields=[_static_filters, _filter])
+
     reference_dt = reference_dt or datetime.now(pytz.utc)
     end_dt = reference_dt - timedelta(hours=delay)
     _LOGGER.debug("Querying with end_dt: %s, that is with delay of %s hrs", end_dt, delay)
@@ -118,6 +141,7 @@ def fetch(
         dimensions: list[str],
         delay: float = 3.0,
         granularity: str = "minute",
+        static_filters: Optional[FilterConf] = None,
         aggregations: Optional[dict] = None,
         post_aggregations: Optional[dict] = None,
         group_by: Optional[list[str]] = None,
@@ -135,6 +159,7 @@ def fetch(
             dimensions: The dimensions to group by
             delay: Added delay to the fetch query from current time.
             granularity: Time bucket to aggregate data by hour, day, minute, etc.
+            static_filters: user defined filters
             aggregations: A map from aggregator name to one of the
                 ``pydruid.utils.aggregators`` e.g., ``doublesum``
             post_aggregations: postaggregations map
@@ -152,6 +177,7 @@ def fetch(
             datasource=datasource,
             dimensions=dimensions,
             filter_pairs=filter_pairs,
+            static_filters=static_filters,
             granularity=granularity,
             hours=hours,
             delay=delay,
@@ -193,6 +219,7 @@ def chunked_fetch(
         dimensions: list[str],
         delay: float = 3.0,
         granularity: str = "minute",
+        static_filter: Optional[FilterConf] = None,
         aggregations: Optional[dict] = None,
         post_aggregations: Optional[dict] = None,
         group_by: Optional[list[str]] = None,
@@ -213,6 +240,7 @@ def chunked_fetch(
             granularity: Time bucket to aggregate data by hour, day, minute, etc.
             aggregations: A map from aggregator name to one of the
                 ``pydruid.utils.aggregators`` e.g., ``doublesum``
+            static_filter: user defined filters
             post_aggregations: postaggregations map
             group_by: List of columns to group by
             pivot: Pivot configuration
@@ -245,6 +273,7 @@ def chunked_fetch(
                     datasource=datasource,
                     dimensions=dimensions,
                     filter_pairs=filter_pairs,
+                    static_filter=static_filter,
                     granularity=granularity,
                     hours=min(chunked_hours, hours - hours_elapsed),
                     delay=delay,
@@ -259,9 +288,7 @@ def chunked_fetch(
         _LOGGER.debug("Fetching data concurrently with %s threads", max_threads)
         with ThreadPoolExecutor(max_workers=max_threads) as executor:
             futures = [executor.submit(self._fetch, **params) for params in qparams]
-            for future in futures:
-                chunked_dfs.append(future.result())
-
+            chunked_dfs.extend(future.result() for future in futures)
         df = pd.concat(chunked_dfs, axis=0, ignore_index=True)
         df["timestamp"] = pd.to_datetime(df["timestamp"]).astype("int64") // 10**6
 

numalogic/udfs/trainer/_druid.py

@@ -120,6 +120,7 @@ def fetch_data(self, payload: TrainerPayload) -> Optional[pd.DataFrame]:
                 datasource=_fetcher_conf.datasource,
                 filter_keys=_stream_conf.composite_keys,
                 filter_values=payload.composite_keys,
+                static_filters=_fetcher_conf.static_filters,
                 dimensions=list(_fetcher_conf.dimensions),
                 delay=self.dataconn_conf.delay_hrs,
                 granularity=_fetcher_conf.granularity,

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "numalogic"
-version = "0.9.1a7"
+version = "0.9.1a8"
 description = "Collection of operational Machine Learning models and tools."
 authors = ["Numalogic Developers"]
 packages = [{ include = "numalogic" }]