Feature Group Compute Framework Integration

Overview

One of mloda's key strengths is its ability to decouple feature definitions from specific computation technologies. This document explains how feature groups integrate with different compute frameworks.

Core Concepts

Compute Framework Specification

Feature groups specify which compute frameworks they support through the compute_framework_rule method:

@classmethod
def compute_framework_rule(cls) -> set[type[ComputeFramework]]:
    """Define the compute frameworks this feature group supports."""
    return {PandasDataFrame}  # Support only Pandas
    # Or return True to support all available compute frameworks

Declaring an Operation Unsupported on a Framework

compute_framework_rule is a static, class-level set: it cannot say "this feature group runs on SQLite in general, but this particular operation is unsupported there." For that, override supports_compute_framework, a per-feature hook evaluated at match time:

@classmethod
def supports_compute_framework(cls, feature_name, options, compute_framework) -> bool:
    """Reject an operation on a specific framework. Default returns True."""
    if compute_framework is SqliteFramework and is_median_op(feature_name, options):
        return False  # median is unsupported on SQLite
    return True

Returning False removes that framework from the candidate set for this feature only:

• If another framework can still run the operation, the matcher routes around the rejected one silently (no error).
• If the only remaining candidate is the rejected framework (for example, the user pinned the feature to it), resolution fails with the standard "No feature groups found" error, which names the near-miss feature group and why it dropped, one line per eliminated candidate under a "Feature group(s) eliminated while matching '...'" block, e.g.

No feature groups found for feature name: 'X'.
Feature group(s) eliminated while matching 'X':
  - MedianFeatureGroup (compute framework pin): pinned compute framework 'SqliteFramework' is not among its supported ['DuckDBFramework', 'PandasDataFrame']
Use resolve_feature(name, options=...) to debug feature resolution.
For troubleshooting guide, see: https://mloda-ai.github.io/mloda/in_depth/troubleshooting/feature-group-resolution-errors/

The rejected framework is named as a near-miss with its reason rather than vanishing into a generic "unknown feature" message. Prefer this hook over raising a generic error from inside calculate_feature: the rejection happens during planning rather than at compute time, and the message is built for you.

The debug inspector resolve_feature(name) reflects the hook too, because it runs the same matcher over the same candidate universe as the engine (it delegates to IdentifyFeatureGroupClass). Its ResolvedFeature result carries supported_compute_frameworks and unsupported_compute_frameworks (evaluated under default options unless you pass options=, see Discover Plugins).

Two distinct gaps both resolve to feature_group=None rather than appearing runnable:

• Unsupported on every framework. The feature matches a group, but supports_compute_framework rejects every installed framework. Resolution fails with the capability error above.
• Uninstalled framework. The only framework a matching group declares is not installed (its backend library is absent, so is_available() is False). The candidate universe drops unavailable frameworks before matching, so the group maps to an empty framework set and resolution fails closed with the ordinary No feature groups found for feature name: '<name>'. error. It does not read as runnable on a framework you cannot actually run.

Declaring capability per subtype

Families with multiple subtypes declare per-backend capability as data instead of a hand-written supports_compute_framework.

The data provider declares the dimension once with SUBTYPES:

class RankFeatureGroup(FeatureChainParserMixin, FeatureGroup):
    SUBTYPES = SubtypeDeclaration(
        key="rank_type",
        parametric_families={"ntile": "N-tile bucketing"},
        supported={"PythonDictFramework": {"dense", "ordinal"}},
    )
    PREFIX_PATTERN = r".*__([\w]+)_rank$"
    PROPERTY_MAPPING = {
        "rank_type": property_spec(
            "Rank subtype.",
            strict=True,
            allowed_values={"dense": "Dense ranking", "ordinal": "Ordinal ranking"},
        ),
    }

Two shapes, enforced at class definition; a half declaration fails at import:

• Shape A: key names a PROPERTY_MAPPING key with an enumerable value space; parametric_families join the universe by family name.
• Shape B (multi-axis families collapsed into one subtype id): declare the universe explicitly with a resolver:

def resolve_window(feature_name: str, options: Options) -> str | None:
    return options.get("window_function")

SUBTYPES = SubtypeDeclaration(universe={"median", "sum"}, resolver=resolve_window)

supported is a sparse per-framework override: frameworks absent from it keep the full universe. A key naming no declared framework is a silent no-op for matching but raises from subtype_support_matrix(), so a typo surfaces as subtype_error in the audit. The derived supports_compute_framework canonicalizes parametric instances (ntile_2 becomes ntile) and gates declared subtypes by the declaration; undeclared subtypes and features without one stay open.

The data steward audits capability via subtype_support_matrix() (supported subtypes per framework from compute_framework_definition(); empty for abstract bases) and get_feature_group_docs() (subtype_key, subtypes, parametric_subtypes, subtype_support, subtype_error). A hand-overridden supports_compute_framework yields no declared matrix; the misfit surfaces as subtype_error.

The data user sees the outcome on resolve_feature(name, options=...): subtype (e.g. ntile_2) and subtype_family (ntile, parametric instances only).

Empty Results

The contract is: a final requested feature must return a schema-bearing result, meaning at least one column. Zero rows is a valid result; zero columns is not. A filter that excludes every row, a join with no matches, or a time window with no events all return a well-typed frame with the right columns and no rows, and mloda passes these through unchanged.

The error fires only when a final requested result carries no schema at all. If calculate_feature produces a result with no columns, ComputeFramework.run_validate_output_features raises:

EmptyResultError: Result carries no schema (no columns): <FeatureGroupClassName>. ...

EmptyResultError is a ValueError subclass and is importable from the public API: from mloda.provider import EmptyResultError. Intermediate feature groups (those whose output feeds another feature group rather than the caller directly) are never subject to this check.

The schema-presence gate

The guard detects a missing schema via the framework's existing ComputeFramework._extract_column_names(self, data) -> set[str]: an empty set means no schema, which is the error condition. Every framework already implements this off schema metadata, so it works on a zero-row frame and costs nothing extra (no row scan, collect, or count). No per-framework opt-in is needed when you implement a new compute framework, as long as _extract_column_names returns the columns for a zero-row frame.

There is one representational caveat. The schema-bearing frameworks (PyArrow, Pandas, Polars, DuckDB, SQLite, Spark, Iceberg) carry their schema as metadata even at zero rows, so a zero-row result keeps its columns and passes. The PythonDict framework represents data as a columnar dict[str, list], where the schema is the set of keys and is present even at zero rows: {"col": []} is a valid schema-bearing zero-row frame, while {} (zero columns) is the only schema-less value. Emptiness is judged purely on schema presence, with no opt-in: a zero-column result raises EmptyResultError uniformly on every framework. One consequence: on a schema-less result (a zero-column frame), column selection returns the result as is, so a misspelled requested column on schema-less data does not produce a "column not found" error.

Filter column validation

_validate_filter_columns skips its column-presence and dtype checks only when the data is the schema-less empty result (in practice PythonDict's {}). Filtering an empty result is a no-op, so neither the column check nor row elimination has anything to do. Data on which the framework cannot see columns but that is not an empty result still fails the missing-filter-column check loudly.

Timezone and unit validation (merge and filter engines)

A custom compute framework's merge and filter engines can opt into the comparison contract, which rejects incompatible timezone/unit combinations in equi-joins, as-of joins, and datetime filter bounds. The guard is opt-in: set provides_column_semantics = True on your BaseMergeEngine / BaseFilterEngine subclass and implement _column_semantics(data, column) to report the column's native semantics. Leave the flag at its default False (for a framework with no temporal intent) and the guard is skipped entirely, so you are never forced to implement the hook. An engine that opts in but forgets the hook raises a clear error rather than silently skipping validation. As-of joins always require _column_semantics regardless of the flag, since ordered time columns are intrinsic to the operation.

Framework-Specific Implementations

Feature groups follow a layered architecture: - Base class defines the interface and common functionality - Framework-specific classes implement the actual calculations

FeatureGroup
  └── BaseFeatureGroup (e.g., ClusteringFeatureGroup)
        ├── PandasImplementation
        ├── PyArrowImplementation
        └── PythonDictFrameworkImplementation

Implementation Pattern

1. Base Class

The base class defines the interface and common functionality:

class MyFeatureGroup(FeatureGroup):
    """Base class for MyFeatureGroup."""

    def input_features(self, options, feature_name):
        # Common logic for extracting input features

    @classmethod
    def calculate_feature(cls, data, features):
        # This will be overridden by framework-specific implementations
        raise NotImplementedError()

2. Framework-Specific Implementation

Each framework-specific implementation: - Specifies which compute frameworks it supports - Implements the calculation logic for that framework

class PandasMyFeatureGroup(MyFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        """Define supported compute frameworks."""
        return {PandasDataFrame}

    @classmethod
    def calculate_feature(cls, data, features):
        """Implement calculation using pandas."""
        # Pandas-specific implementation

Framework Selection Process

When a feature is requested:

1. The system identifies the appropriate feature group
2. It checks which compute frameworks are supported by:
3. The feature definition
4. The feature group
5. The mloda request
6. It selects a compatible compute framework
7. It uses the framework-specific implementation for calculations

Data Transformation

When data needs to move between compute frameworks:

1. The transform method converts data between frameworks
2. Each framework defines how to transform data to and from other frameworks
3. The system automatically handles these transformations when needed

For more details on how data transformation works between compute frameworks, see Framework Transformers.

Example

For a clustering feature group:

# Base class (framework-agnostic)
class ClusteringFeatureGroup(FeatureGroup):
    def input_features(self, options, feature_name):
        # Extract source features from feature name

    @classmethod
    def calculate_feature(cls, data, features):
        # This will be overridden by framework-specific implementations

# Pandas implementation
class PandasClusteringFeatureGroup(ClusteringFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        return {PandasDataFrame}

    @classmethod
    def calculate_feature(cls, data, features):
        # Pandas-specific clustering implementation

# PyArrow implementation
class PyArrowClusteringFeatureGroup(ClusteringFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        return {PyArrowTable}

    @classmethod
    def calculate_feature(cls, data, features):
        # PyArrow-specific clustering implementation

For an aggregated feature group with Polars support:

# Base class (framework-agnostic)
class AggregatedFeatureGroup(FeatureGroup):
    def input_features(self, options, feature_name):
        # Extract source features from feature name

    @classmethod
    def calculate_feature(cls, data, features):
        # This will be overridden by framework-specific implementations

# Polars Lazy implementation
class PolarsLazyAggregatedFeatureGroup(AggregatedFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        return {PolarsLazyDataFrame}

    @classmethod
    def calculate_feature(cls, data, features):
        # Polars lazy-specific aggregation implementation
        # Uses lazy evaluation for query optimization

Note that Polars supports both eager (PolarsDataFrame) and lazy (PolarsLazyDataFrame) evaluation modes, allowing you to choose the appropriate strategy based on your performance requirements.

For an analytical feature group with DuckDB support:

# Base class (framework-agnostic)
class AnalyticalFeatureGroup(FeatureGroup):
    def input_features(self, options, feature_name):
        # Extract source features from feature name

    @classmethod
    def calculate_feature(cls, data, features):
        # This will be overridden by framework-specific implementations

# DuckDB implementation
class DuckDBAnalyticalFeatureGroup(AnalyticalFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        return {DuckDBFramework}

    @classmethod
    def calculate_feature(cls, data, features):
        # DuckDB-specific analytical implementation
        # Uses SQL-like operations for complex analytics
        # Example: data.aggregate("column", "sum").df()
        return data.aggregate("value_column", "sum")

Important: DuckDB feature groups require a connection object to be available. The framework will automatically handle connection management, but ensure your data access collection includes the necessary connection information.

For a distributed processing feature group with Spark support:

# Base class (framework-agnostic)
class DistributedFeatureGroup(FeatureGroup):
    def input_features(self, options, feature_name):
        # Extract source features from feature name

    @classmethod
    def calculate_feature(cls, data, features):
        # This will be overridden by framework-specific implementations

# Spark implementation
class SparkDistributedFeatureGroup(DistributedFeatureGroup):
    @classmethod
    def compute_framework_rule(cls):
        return {SparkFramework}

    @classmethod
    def calculate_feature(cls, data, features):
        # Spark-specific distributed processing implementation
        # Uses Spark DataFrame operations for scalable processing
        # Example: data.groupBy("category").agg({"value": "sum"})
        return data.groupBy("category_column").agg({"value_column": "sum"})

Important: Spark feature groups require PySpark installation and Java 8+ environment with JAVA_HOME configured. The framework can auto-create a local SparkSession if none is provided, but for production use, you should provide a configured SparkSession through the data access collection. Spark uses its own distributed processing capabilities instead of mloda's framework inherent multiprocessing.

SQL Relation Helpers (DuckDB / SQLite)

The DuckDB and SQLite frameworks expose a relation object (DuckdbRelation, SqliteRelation) inside calculate_feature. Both share the same helper surface so a feature group can be written once against either.

Reading column types

The .types property returns column types aligned with .columns. The element type differs by backend:

• DuckdbRelation.types returns DuckDB-native dtype objects.
• SqliteRelation.types returns PyArrow pa.DataType objects (from propagated hints, falling back to SQLite affinity inference).

relation.columns   # ["user_id", "amount"]
relation.types     # backend-specific dtype objects, same order as columns

Window functions

with_row_number appends a ROW_NUMBER() column; window appends an arbitrary window expression. Both quote every identifier and raise ValueError if the new alias collides with an existing column.

from mloda_plugins.compute_framework.base_implementations.sql.sql_window import (
    OrderBy,
    WindowFrame,
    Preceding,
    CurrentRow,
)

# ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts)
ranked = relation.with_row_number(
    "rn",
    partition_by=["user_id"],
    order_by=[OrderBy("ts")],
)

# SUM(amount) OVER (PARTITION BY user_id ORDER BY ts ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
rolling = relation.window(
    "SUM(amount)",
    "amount_rolling",
    partition_by=["user_id"],
    order_by=[OrderBy("ts")],
    frame=WindowFrame(kind="rows", start=Preceding(2), end=CurrentRow()),
)

order_by accepts plain column-name strings or OrderBy(column, descending=..., nulls="first"|"last"). The func passed to window is inlined verbatim as raw SQL, so never build it from user-controlled input.

!!! warning "SQLite version requirement" On SQLite, with_row_number and window require SQLite >= 3.28.0; using NULLS placement in order_by additionally requires SQLite >= 3.30.0. Both raise ValueError on older runtimes. DuckDB has no such gate.