Skip to content

API: preprocessing.drop_and_missing

skyulf.preprocessing.drop_and_missing

Drop / missing-value nodes package.

Split from a single 409-LOC module into per-node files

_common.py — shared helpers (y-index sync, subset normalization) deduplicate.py — Deduplicate drop_columns.py — DropMissingColumns drop_rows.py — DropMissingRows missing_indicator.py — MissingIndicator

All public names are re-exported here so existing imports such as from skyulf.preprocessing.drop_and_missing import DeduplicateCalculator continue to work unchanged.

DeduplicateApplier

Bases: BaseApplier

Drop duplicate rows, keeping y aligned with the survivors.

Both engines must agree on which rows survive. y is filtered by kept positions, never by label — a label-based .loc returns every row matching a duplicated index label. On polars the dedup key has to be resolved to the real columns before the tracking row index is added, because unique(subset=None) would include that always-unique index and defeat deduplication entirely. keep="none" drops every member of a duplicate group and is spelled False for pandas but "none" for polars.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class DeduplicateApplier(BaseApplier):
    """Drop duplicate rows, keeping ``y`` aligned with the survivors.

    Both engines must agree on which rows survive. ``y`` is filtered by kept
    *positions*, never by label — a label-based ``.loc`` returns every row
    matching a duplicated index label. On polars the dedup key has to be
    resolved to the real columns before the tracking row index is added,
    because ``unique(subset=None)`` would include that always-unique index and
    defeat deduplication entirely. ``keep="none"`` drops every member of a
    duplicate group and is spelled ``False`` for pandas but ``"none"`` for
    polars.
    """

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Require complete duplicate groups before choosing which rows survive."""
        if engine not in ("pandas", "polars"):
            return None
        return ExecutionCapability(engine, "apply", "local", "filter", "global")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch to the engine-specific dedup, forwarding ``(X, y)`` only when ``y`` exists."""
        # Note: dedup must propagate row drops to y, so we route X+y as a tuple
        # through apply_dual_engine which handles unpack/pack.
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _dedup_apply_polars, "pandas": _dedup_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific dedup, forwarding (X, y) only when y exists.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
86
87
88
89
90
91
92
93
94
95
@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific dedup, forwarding ``(X, y)`` only when ``y`` exists."""
    # Note: dedup must propagate row drops to y, so we route X+y as a tuple
    # through apply_dual_engine which handles unpack/pack.
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _dedup_apply_polars, "pandas": _dedup_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Require complete duplicate groups before choosing which rows survive.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
79
80
81
82
83
84
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Require complete duplicate groups before choosing which rows survive."""
    if engine not in ("pandas", "polars"):
        return None
    return ExecutionCapability(engine, "apply", "local", "filter", "global")

DeduplicateCalculator

Bases: BaseCalculator

Resolve dedup config into an artifact.

The node is flagged learns_from_data because which rows survive depends on the data, but fit itself only reads config.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@NodeRegistry.register("Deduplicate", DeduplicateApplier)
@node_meta(
    id="Deduplicate",
    name="Deduplicate",
    category="Data Operations",
    description="Drop duplicate rows.",
    params={"subset": [], "keep": "first"},
    learns_from_data=True,
)
class DeduplicateCalculator(BaseCalculator):
    """Resolve dedup config into an artifact.

    The node is flagged ``learns_from_data`` because which rows survive depends
    on the data, but ``fit`` itself only reads ``config``.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: rows are dropped, columns are preserved."""
        # Deduplication removes rows; column set is preserved.
        return input_schema

    def fit(self, df: Any, config: dict[str, Any]) -> DeduplicateArtifact:
        """Carry the ``subset`` and ``keep`` policy into the artifact."""
        return {
            "type": "deduplicate",
            "subset": config.get("subset"),
            "keep": config.get("keep", "first"),
        }

fit(df, config)

Carry the subset and keep policy into the artifact.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
121
122
123
124
125
126
127
def fit(self, df: Any, config: dict[str, Any]) -> DeduplicateArtifact:
    """Carry the ``subset`` and ``keep`` policy into the artifact."""
    return {
        "type": "deduplicate",
        "subset": config.get("subset"),
        "keep": config.get("keep", "first"),
    }

infer_output_schema(input_schema, config)

Pass the input schema through: rows are dropped, columns are preserved.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/deduplicate.py 
114
115
116
117
118
119
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: rows are dropped, columns are preserved."""
    # Deduplication removes rows; column set is preserved.
    return input_schema

DropMissingColumnsApplier

Bases: BaseApplier

Drop the columns the calculator selected, skipping names no longer on the frame.

The selection is already baked into params; this half is a pure drop and leaves y alone, so both engine paths are identical by construction.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class DropMissingColumnsApplier(BaseApplier):
    """Drop the columns the calculator selected, skipping names no longer on the frame.

    The selection is already baked into ``params``; this half is a pure drop
    and leaves ``y`` alone, so both engine paths are identical by construction.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved drop set; threshold metadata no longer controls apply."""
        local_state_fields(raw, "drop_missing_columns", {"type", "columns_to_drop", "threshold"})
        _columns(raw["columns_to_drop"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local saved apply context without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        DropMissingColumnsApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the column drop to the pandas or polars path; ``y`` passes through."""
        return apply_dual_engine(
            X,
            params,
            {"polars": _drop_missing_cols_apply_polars, "pandas": _drop_missing_cols_apply_pandas},
        )

apply(X, _y, params)

Dispatch the column drop to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
50
51
52
53
54
55
56
57
@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the column drop to the pandas or polars path; ``y`` passes through."""
    return apply_dual_engine(
        X,
        params,
        {"polars": _drop_missing_cols_apply_polars, "pandas": _drop_missing_cols_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe local saved apply context without granting worker execution.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
42
43
44
45
46
47
48
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe local saved apply context without granting worker execution."""
    if engine not in ("pandas", "polars"):
        return None
    DropMissingColumnsApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect the saved drop set; threshold metadata no longer controls apply.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
35
36
37
38
39
40
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved drop set; threshold metadata no longer controls apply."""
    local_state_fields(raw, "drop_missing_columns", {"type", "columns_to_drop", "threshold"})
    _columns(raw["columns_to_drop"])
    return raw

DropMissingColumnsCalculator

Bases: BaseCalculator

Select columns to drop: an explicit list unioned with those over the missing-% threshold.

The threshold half is data-dependent, so fit runs an engine-specific path and the two must agree on the missing count. They do not for free: polars' null_count() ignores NaN where pandas' isna() counts it, so float columns add an explicit NaN count on the polars side.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@NodeRegistry.register("DropMissingColumns", DropMissingColumnsApplier)
@node_meta(
    id="DropMissingColumns",
    name="Drop Missing Columns",
    category="Cleaning",
    description="Drop columns that exceed missing value threshold.",
    params={"missing_threshold": 50},
    learns_from_data=True,
)
class DropMissingColumnsCalculator(BaseCalculator):
    """Select columns to drop: an explicit list unioned with those over the missing-% threshold.

    The threshold half is data-dependent, so ``fit`` runs an engine-specific
    path and the two must agree on the missing count. They do not for free:
    polars' ``null_count()`` ignores NaN where pandas' ``isna()`` counts it, so
    float columns add an explicit NaN count on the polars side.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Predict the surviving columns, or ``None`` when a threshold makes that data-dependent.

        Without a positive threshold the result is exactly the input minus the
        explicit list; with one, which columns go depends on the data and
        callers must fall back to runtime introspection.
        """
        # Threshold path is data-dependent; only predictable when the user
        # supplied an explicit column list and no positive threshold.
        if _resolve_threshold(config.get("missing_threshold")) is not None:
            return None
        explicit = config.get("columns", []) or []
        return input_schema if not explicit else input_schema.drop(explicit)

    def fit(self, df: Any, config: dict[str, Any]) -> DropMissingColumnsArtifact:
        """Compute the drop set on the frame's own engine and return it as the artifact."""
        return cast(
            DropMissingColumnsArtifact,
            fit_dual_engine(
                df,
                config,
                {"polars": _drop_missing_cols_fit_polars, "pandas": _drop_missing_cols_fit_pandas},
            ),
        )

fit(df, config)

Compute the drop set on the frame's own engine and return it as the artifact.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
163
164
165
166
167
168
169
170
171
172
def fit(self, df: Any, config: dict[str, Any]) -> DropMissingColumnsArtifact:
    """Compute the drop set on the frame's own engine and return it as the artifact."""
    return cast(
        DropMissingColumnsArtifact,
        fit_dual_engine(
            df,
            config,
            {"polars": _drop_missing_cols_fit_polars, "pandas": _drop_missing_cols_fit_pandas},
        ),
    )

infer_output_schema(input_schema, config)

Predict the surviving columns, or None when a threshold makes that data-dependent.

Without a positive threshold the result is exactly the input minus the explicit list; with one, which columns go depends on the data and callers must fall back to runtime introspection.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_columns.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema | None:
    """Predict the surviving columns, or ``None`` when a threshold makes that data-dependent.

    Without a positive threshold the result is exactly the input minus the
    explicit list; with one, which columns go depends on the data and
    callers must fall back to runtime introspection.
    """
    # Threshold path is data-dependent; only predictable when the user
    # supplied an explicit column list and no positive threshold.
    if _resolve_threshold(config.get("missing_threshold")) is not None:
        return None
    explicit = config.get("columns", []) or []
    return input_schema if not explicit else input_schema.drop(explicit)

DropMissingRowsApplier

Bases: BaseApplier

Drop rows exceeding the configured missingness policy, keeping y in sync.

This node changes the row count, so the pandas and polars paths must drop the same rows and filter y by the identical kept positions. The pandas path therefore computes its keep mask positionally: label-based .loc returns every row matching a duplicated index label, which desynchronises y from the cleaned X. A subset that is empty or names only nonexistent columns collapses to None, which both engines read as "check every column" rather than as a no-op.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class DropMissingRowsApplier(BaseApplier):
    """Drop rows exceeding the configured missingness policy, keeping ``y`` in sync.

    This node changes the row count, so the pandas and polars paths must drop
    the same rows *and* filter ``y`` by the identical kept positions. The
    pandas path therefore computes its keep mask positionally: label-based
    ``.loc`` returns every row matching a duplicated index label, which
    desynchronises ``y`` from the cleaned ``X``. A ``subset`` that is empty or
    names only nonexistent columns collapses to ``None``, which both engines
    read as "check every column" rather than as a no-op.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved policy without filtering data or converting native values."""
        local_state_fields(
            raw, "drop_missing_rows", {"type", "subset", "how", "threshold", "missing_threshold"}
        )
        _validate_how(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe native filtering; ordinary prediction skips this training policy."""
        if engine not in ("pandas", "polars"):
            return None
        DropMissingRowsApplier.validate_inference_state(state)
        subset = state["subset"]
        if subset is not None and not isinstance(subset, (str, list, tuple, dict, set, frozenset)):
            return None
        threshold = state["threshold"]
        if threshold is None:
            threshold = state["missing_threshold"]
        if threshold is not None and not isinstance(threshold, (Number, np.bool_)):
            # Array/Series thresholds can depend on request positions or indexes.
            return None
        return ExecutionCapability(engine, "apply", "local", "filter", "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch to the engine-specific dropna, forwarding ``(X, y)`` only when ``y`` exists."""
        _validate_how(params)
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _drop_missing_rows_apply_polars, "pandas": _drop_missing_rows_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific dropna, forwarding (X, y) only when y exists.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
155
156
157
158
159
160
161
162
163
@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific dropna, forwarding ``(X, y)`` only when ``y`` exists."""
    _validate_how(params)
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _drop_missing_rows_apply_polars, "pandas": _drop_missing_rows_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe native filtering; ordinary prediction skips this training policy.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe native filtering; ordinary prediction skips this training policy."""
    if engine not in ("pandas", "polars"):
        return None
    DropMissingRowsApplier.validate_inference_state(state)
    subset = state["subset"]
    if subset is not None and not isinstance(subset, (str, list, tuple, dict, set, frozenset)):
        return None
    threshold = state["threshold"]
    if threshold is None:
        threshold = state["missing_threshold"]
    if threshold is not None and not isinstance(threshold, (Number, np.bool_)):
        # Array/Series thresholds can depend on request positions or indexes.
        return None
    return ExecutionCapability(engine, "apply", "local", "filter", "row")

validate_inference_state(raw) staticmethod

Inspect the saved policy without filtering data or converting native values.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
129
130
131
132
133
134
135
136
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved policy without filtering data or converting native values."""
    local_state_fields(
        raw, "drop_missing_rows", {"type", "subset", "how", "threshold", "missing_threshold"}
    )
    _validate_how(raw)
    return raw

DropMissingRowsCalculator

Bases: BaseCalculator

Resolve drop-missing-rows config into an artifact; nothing is learned from data.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@NodeRegistry.register("DropMissingRows", DropMissingRowsApplier)
@node_meta(
    id="DropMissingRows",
    name="Drop Missing Rows",
    category="Cleaning",
    description="Drop rows containing missing values in specified columns.",
    params={"subset": [], "how": "any", "missing_threshold": None},
    learns_from_data=False,
)
class DropMissingRowsCalculator(BaseCalculator):
    """Resolve drop-missing-rows config into an artifact; nothing is learned from data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: rows are dropped, columns are preserved."""
        # Drops rows; column set is preserved.
        return input_schema

    def fit(self, df: Any, config: dict[str, Any]) -> DropMissingRowsArtifact:
        """Carry the dropna policy into the artifact.

        Reads only ``config``, so it takes ``df`` directly and needs no
        ``@fit_method`` unpacking of ``(X, y)``. Both engines apply the policy
        in the same precedence: an absolute ``threshold`` outranks the
        percentage ``missing_threshold``, which outranks ``how``.
        """
        return {
            "type": "drop_missing_rows",
            "subset": config.get("subset"),
            "how": _validate_how(config),
            "threshold": config.get("threshold"),
            "missing_threshold": config.get("missing_threshold"),
        }

fit(df, config)

Carry the dropna policy into the artifact.

Reads only config, so it takes df directly and needs no @fit_method unpacking of (X, y). Both engines apply the policy in the same precedence: an absolute threshold outranks the percentage missing_threshold, which outranks how.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def fit(self, df: Any, config: dict[str, Any]) -> DropMissingRowsArtifact:
    """Carry the dropna policy into the artifact.

    Reads only ``config``, so it takes ``df`` directly and needs no
    ``@fit_method`` unpacking of ``(X, y)``. Both engines apply the policy
    in the same precedence: an absolute ``threshold`` outranks the
    percentage ``missing_threshold``, which outranks ``how``.
    """
    return {
        "type": "drop_missing_rows",
        "subset": config.get("subset"),
        "how": _validate_how(config),
        "threshold": config.get("threshold"),
        "missing_threshold": config.get("missing_threshold"),
    }

infer_output_schema(input_schema, config)

Pass the input schema through: rows are dropped, columns are preserved.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/drop_rows.py 
178
179
180
181
182
183
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: rows are dropped, columns are preserved."""
    # Drops rows; column set is preserved.
    return input_schema

MissingIndicatorApplier

Bases: BaseApplier

Add a binary <col><flag_suffix> column beside each configured column.

Additive, not replacing: the original columns survive. Both engines must flag the same cells, which needs care on polars — it stores float NaN distinctly from null where pandas' isna() treats both as missing, so float columns get an explicit NaN test. Columns absent from the frame are skipped. A generated flag name matching an existing column raises ValueError; input values are never overwritten by flags.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class MissingIndicatorApplier(BaseApplier):
    """Add a binary ``<col><flag_suffix>`` column beside each configured column.

    Additive, not replacing: the original columns survive. Both engines must
    flag the same cells, which needs care on polars — it stores float NaN
    distinctly from null where pandas' ``isna()`` treats both as missing, so
    float columns get an explicit NaN test. Columns absent from the frame are
    skipped. A generated flag name matching an existing column raises
    ``ValueError``; input values are never overwritten by flags.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect learned flag selection without detecting missing values again."""
        local_state_fields(raw, "missing_indicator", {"type", "columns", "flag_suffix"})
        columns = _columns(raw["columns"])
        if type(raw["flag_suffix"]) is not str or not raw["flag_suffix"]:
            raise ValueError("Fitted missing-indicator suffix must be a nonempty string.")
        validate_generated_column_names(
            columns,
            (f"{column}{raw['flag_suffix']}" for column in columns),
            node_name="MissingIndicator",
        )
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local saved apply context without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        MissingIndicatorApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the flag creation to the pandas or polars path; ``y`` passes through."""
        return apply_dual_engine(
            X,
            params,
            {"polars": _missing_indicator_apply_polars, "pandas": _missing_indicator_apply_pandas},
        )

apply(X, _y, params)

Dispatch the flag creation to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
 98
 99
100
101
102
103
104
105
@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the flag creation to the pandas or polars path; ``y`` passes through."""
    return apply_dual_engine(
        X,
        params,
        {"polars": _missing_indicator_apply_polars, "pandas": _missing_indicator_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe local saved apply context without granting worker execution.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
90
91
92
93
94
95
96
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe local saved apply context without granting worker execution."""
    if engine not in ("pandas", "polars"):
        return None
    MissingIndicatorApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect learned flag selection without detecting missing values again.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect learned flag selection without detecting missing values again."""
    local_state_fields(raw, "missing_indicator", {"type", "columns", "flag_suffix"})
    columns = _columns(raw["columns"])
    if type(raw["flag_suffix"]) is not str or not raw["flag_suffix"]:
        raise ValueError("Fitted missing-indicator suffix must be a nonempty string.")
    validate_generated_column_names(
        columns,
        (f"{column}{raw['flag_suffix']}" for column in columns),
        node_name="MissingIndicator",
    )
    return raw

MissingIndicatorCalculator

Bases: BaseCalculator

Choose which columns get a missing flag, learning the set from data when none is named.

An explicit columns list is filtered down to names present on the frame; without one, every column that actually contains a missing value is flagged, so the artifact depends on the training data.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@NodeRegistry.register("MissingIndicator", MissingIndicatorApplier)
@node_meta(
    id="MissingIndicator",
    name="Missing Indicator",
    category="Feature Engineering",
    description="Create binary indicators for missing values.",
    params={"columns": [], "flag_suffix": _DEFAULT_FLAG_SUFFIX},
    learns_from_data=True,
)
class MissingIndicatorCalculator(BaseCalculator):
    """Choose which columns get a missing flag, learning the set from data when none is named.

    An explicit ``columns`` list is filtered down to names present on the
    frame; without one, every column that actually contains a missing value is
    flagged, so the artifact depends on the training data.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Predict the added flag columns, or ``None`` when the set is data-dependent.

        With an explicit list the output is the input plus one ``int64``
        ``<col><flag_suffix>`` column for each source present in the input;
        otherwise which columns hold missing values decides, and callers
        must introspect at runtime.
        """
        # Adds one int64 (0/1) column "<col><flag_suffix>" per indicator
        # column. Only predictable when the user supplied an explicit column
        # list; otherwise the set depends on which columns actually contain
        # missing values.
        explicit = config.get("columns") or []
        if not explicit:
            return None
        suffix = config.get("flag_suffix") or _DEFAULT_FLAG_SUFFIX
        new_schema = input_schema
        for col in explicit:
            if col in input_schema:
                new_schema = new_schema.add(f"{col}{suffix}", "int64")
        return new_schema

    def fit(self, df: Any, config: dict[str, Any]) -> MissingIndicatorArtifact:
        """Compute the flagged column set on the frame's own engine and return it."""
        return cast(
            MissingIndicatorArtifact,
            fit_dual_engine(
                df,
                config,
                {"polars": _missing_indicator_fit_polars, "pandas": _missing_indicator_fit_pandas},
            ),
        )

fit(df, config)

Compute the flagged column set on the frame's own engine and return it.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
191
192
193
194
195
196
197
198
199
200
def fit(self, df: Any, config: dict[str, Any]) -> MissingIndicatorArtifact:
    """Compute the flagged column set on the frame's own engine and return it."""
    return cast(
        MissingIndicatorArtifact,
        fit_dual_engine(
            df,
            config,
            {"polars": _missing_indicator_fit_polars, "pandas": _missing_indicator_fit_pandas},
        ),
    )

infer_output_schema(input_schema, config)

Predict the added flag columns, or None when the set is data-dependent.

With an explicit list the output is the input plus one int64 <col><flag_suffix> column for each source present in the input; otherwise which columns hold missing values decides, and callers must introspect at runtime.

Source code in skyulf-core/skyulf/preprocessing/drop_and_missing/missing_indicator.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema | None:
    """Predict the added flag columns, or ``None`` when the set is data-dependent.

    With an explicit list the output is the input plus one ``int64``
    ``<col><flag_suffix>`` column for each source present in the input;
    otherwise which columns hold missing values decides, and callers
    must introspect at runtime.
    """
    # Adds one int64 (0/1) column "<col><flag_suffix>" per indicator
    # column. Only predictable when the user supplied an explicit column
    # list; otherwise the set depends on which columns actually contain
    # missing values.
    explicit = config.get("columns") or []
    if not explicit:
        return None
    suffix = config.get("flag_suffix") or _DEFAULT_FLAG_SUFFIX
    new_schema = input_schema
    for col in explicit:
        if col in input_schema:
            new_schema = new_schema.add(f"{col}{suffix}", "int64")
    return new_schema