Skip to content

SQL-Native Architecture

Pilz is not fully SQL-native. Training data loading and per-node sampling run natively in Polars, and freshly trained (condensed) models are scored by a native Polars tree walk. DuckDB/SQL is still used for the deployment rules, legacy cut-model scoring and the multi-class label combination. This page describes which part runs where.

Data Loading

Training and test data is loaded from CSV or Parquet files with Polars:

def get_df_from_files(self, files: list[str]) -> pl.DataFrame:
    ...
    if files[0].endswith(".csv"):
        df = pl.concat(
            [pl.read_csv(f, schema_overrides=schema_overrides) for f in files]
        )
    elif files[0].endswith(".parquet"):
        df = pl.concat([pl.read_parquet(f) for f in files]).pipe(
            _coerce_feature_dtypes, self.dc.features
        )
    else:
        raise ValueError("Unsupported file type")

The cached training frame is shuffled once at load time, so a per-node filter does not need an ORDER BY RANDOM() query:

def get_cached_train_df(self) -> pl.DataFrame:
    if self.df_train_cach is None:
        self.df_train_cach = self.get_df_from_files(self.train_files).sample(
            fraction=1.0, shuffle=True
        )
    return self.df_train_cach

Per-node sampling is a lazy Polars filter followed by .head(max_eval_fit) — no DuckDB and no ORDER BY RANDOM():

def _get_pl_train_df(self, full_filters: Boolean, max_eval_fit) -> pl.DataFrame:
    cached = self.get_cached_train_df()
    expr = expr_to_polars_expr(
        full_filters, schema=cached.schema, cache=self._pl_expr_cache
    )
    return cached.lazy().filter(expr).head(max_eval_fit).collect()

Filter Translation

Every decision in Pilz is a small boolean expression AST defined in pilz/model/expr.py (Cmp, IsNull, Not, And, Or, Bool). SymPy is gone; Boolean is only a compatibility alias for the new Expr union. The renderer walks the tree recursively and emits a SQL WHERE clause:

class SympyToSqlHelper:
    @staticmethod
    def to_sql_where(expr: Expr) -> str:
        return SympyToSqlHelper._to_sql_impl(expr)

    @staticmethod
    def _to_sql_impl(expr: Expr) -> str:
        if isinstance(expr, And):
            conv = [SympyToSqlHelper._to_sql_impl(arg) for arg in expr.args]
            return f"( {' AND '.join(conv)} )"
        if isinstance(expr, Or):
            conv = [SympyToSqlHelper._to_sql_impl(arg) for arg in expr.args]
            return f"( {' OR '.join(conv)} )"
        if isinstance(expr, Not):
            return f" NOT {SympyToSqlHelper._to_sql_impl(expr.arg)} "
        if isinstance(expr, IsNull):
            op = "IS NOT NULL" if expr.negated else "IS NULL"
            return f" \"{expr.col}\" {op} "
        if isinstance(expr, Cmp):
            op = "<>" if expr.op == "!=" else expr.op
            return f" \"{expr.col}\"{op}{SympyToSqlHelper._render_value(expr.value)} "
        if isinstance(expr, Bool):
            return "TRUE" if expr.value else "FALSE"
        raise TypeError(f"Unknown expression type: {type(expr)} ({expr})")

    @staticmethod
    def _render_value(value) -> str:
        if isinstance(value, str):
            return f"'{value}'"
        return f"{float(value)}"

Null checks are their own AST node (IsNull), not an Eq/Ne against a sentinel. Values render with _render_value: strings quoted, numbers always with a decimal point.

Filters are created via SympyContainer, which wraps a feature name, operator, and value and builds the corresponding AST:

class SympyContainer(BaseModel):
    feat_name: str
    operator: Literal["=", "<", "<=", ">", ">=", "!=", "not in", "in", "between", "is null", "is not null"]
    value: str | int | float | list[str] | list[int] | list[float] | None
    assumptions: dict[str, bool]

    def get_sympy(self) -> Expr:
        col = self.feat_name
        match self.operator:
            case "=":
                return Cmp("=", col, self.value)
            case "in":
                return make_or([Cmp("=", col, value) for value in self.value])
            case "between":
                return make_and([Cmp(">", col, self.value[0]), Cmp("<=", col, self.value[1])])
            case "is null":
                return IsNull(col, negated=False)
            # ...

class Filter:
    def __init__(self, combine: Expr):
        self.combine = combine
        self._sql_cache: dict[bool, str] = {}

    def sql(self, do_invert: bool = False) -> str:
        cached = self._sql_cache.get(do_invert)
        if cached is None:
            combine = make_not(self.combine) if do_invert else self.combine
            cached = SympyToSqlHelper.to_sql_where(combine)
            self._sql_cache[do_invert] = cached
        return cached

Evaluation and Scoring

Scoring picks a different engine per tree. Freshly trained models carry structured Cond conditions and are scored by a native Polars tree walk (_pilz_score_walk); only legacy cut-models are rendered to SQL and run through DuckDB. The per-tree columns are then combined:

def _tree_score_df(self, pilze: Pilze, max_parallel_where: int) -> pl.DataFrame:
    source_df = self.get_cached_eval_df()
    df_sum = None
    for name, pilz in pilze.pilze.items():
        name = name.replace("-", "_")
        if pilz.has_cond:
            # Native Polars tree walk over the structured conditions
            df = self._pilz_score_walk(pilz, source_df, name=name)
        elif len(pilz.spore) < max_parallel_where:
            case_sql = pilz.get_sql(name)
            df = self._get_pl_eval_df(col_sql=case_sql, df=source_df)
        else:
            # Legacy cut-models: split large models into batched SQL queries
            where_cases = pilz.get_split_sql(max_parallel_where=max_parallel_where)
            df = (
                pl.concat(
                    [
                        self._get_pl_eval_df(col_sql=sub_sql, df=source_df)
                        for sub_sql in where_cases
                    ],
                    how="horizontal",
                )
                .sum_horizontal()
                .alias(name)
                .to_frame()
            )
        df_sum = df_sum.hstack(df) if df_sum is not None else df

    return df_sum

The per-tree columns are combined by _combine_tree_scores. mean takes mean_horizontal; max is not max_horizontal — it picks the signed value with the greatest absolute magnitude:

if same_target_pilz_comb_method == "mean":
    return (df_sum.mean_horizontal()).alias(target)

if same_target_pilz_comb_method == "max":
    alle_spalten = df_sum.columns
    return (
        df_sum.select(
            pl.concat_list(alle_spalten).list.get(
                pl.concat_list(pl.col(alle_spalten).abs()).list.arg_max()
            )
        )
        .to_series()
        .alias(target)
    )

raise ValueError(f"{same_target_pilz_comb_method} method not implemented")

get_eval_sr wraps the two steps for a single target.

Model SQL Output

A trained Pilz model can generate SQL. get_where_sql() renders each spore via spore.get_cuts() (the structured cond, falling back to legacy cut), and get_sql emits no ELSE:

class Pilz(BaseModel):
    spore: list[Spore]
    target: str | int

    def get_sql(self, res_name: str) -> str:
        return "CASE \n" + "\n".join(self.get_where_sql()) + f"END AS {res_name}"

    def get_split_sql(self, max_parallel_where: int) -> list[str]:
        where_cases = self.get_where_sql()
        return [
            "CASE \n" + "\n".join(batch) + f"\nELSE CAST(0 AS DOUBLE)\n END AS res_{i}"
            for i, batch in enumerate(batched(where_cases, max_parallel_where))
        ]

    def get_where_sql(self) -> list[str]:
        return [
            f"  WHEN {' AND '.join(spore.get_cuts()) or 'TRUE'} THEN CAST({spore.score} AS DOUBLE) "
            for spore in self.spore
        ]

Each spore becomes one WHEN clause. get_sql has no ELSE (unmatched rows yield NULL); get_split_sql adds ELSE CAST(0 AS DOUBLE) per batch so the batches can be summed. For models with many spores it batches them into parallel SQL queries to stay within database limits.

Ensemble Combination

Pilze is a plain container with only the fields pilze and target; it has no get_sql. For ensembles the per-tree score columns are combined at runtime by _combine_tree_scores using the configured same_target_pilz_comb_method (mean or max), as shown above. Iterate the trees yourself if you want to emit SQL for each one:

sql = {f"res_{name}": pilz.get_sql(f"res_{name}") for name, pilz in pilze.pilze.items()}

For multi-class problems the per-target scores are combined into one prediction either by thresholding (different_target_pilz_comb_method="youden", with per-tree Youden thresholds derived on the fly from the stored leaf counts) or by taking the highest score ("max").

Summary

Component Engine / Role
Data loading Polars read_csv / read_parquet, cached and shuffled once
Filter translation expr.py AST → SQL WHERE clause (and → Polars expressions)
Training sampling Polars lazy filter + .head(max_eval_fit)
Condensed-model scoring Native Polars tree walk (_pilz_score_walk)
Legacy cut-model scoring Pilz.get_sql / get_split_sql via DuckDB
Ensemble combination _combine_tree_scores (mean or max)
Model output CASE WHEN cond THEN CAST(score AS DOUBLE) (no ELSE on get_sql)

Next Steps