Refactor profiler column metrics into an extensible registry

[IvannKurchenko]

Changes

Sets up an extension point for the profiler by moving inline column-metric aggregation in DQProfiler._profile into a registry-based system. Users can now register their own column metric functions via the @register_profile_column_metric() decorator, mirroring the existing register_rule / register_profile_builder extension patterns.

The three existing metrics (count_non_null, count_distinct, empty_count) are moved into this registry with no behavioural change. _build_column_metrics builds the aggregation from whatever is registered.

Rationale for landing the refactoring without new metrics

The profiling pipeline is: column metrics → profile builder → check. Adding a new metric only adds value once a builder consumes it and generates a check. Two paths were considered for #1067:

1. Add percentile metrics (p10/p90) with a matching profile builder that emits is_aggr_not_less_than / is_aggr_not_greater_than checks. But these checks are mostly useful for measurement data (revenue, sales amount, latency, temperature) and not meaningful for keys or categorical columns. Applying them indiscriminately would generate false positives. Selective, purpose-aware application is tracked in [FEATURE]: Profile classification support #1343.
2. Land the refactoring only, expose the extension point, and let users register the metrics they need. This keeps the profiler flexible without shipping metrics that don't yet have a purpose-fit builder.

This PR takes option 2. New built-in metrics can be added later, together with the specific builder that consumes them, once the classification work in #1343 makes selective application safe.

What changed

• New PROFILE_COLUMN_METRIC_REGISTRY and register_profile_column_metric decorator in profiler/profiler_column_metrics.py
• Existing metrics (count_non_null, count_distinct, empty_count) moved into the registry
• DQProfiler._profile refactored: inline aggregation extracted into _build_column_metrics, which iterates the registry
• is_text helper moved from profile_builder.py to profiler/common.py (now used across modules)

Linked issues

Relates to #1067, related to #1343

Tests

• added unit tests
• manually tested
• added integration tests
• added end-to-end tests
• added performance tests

Unit tests cover: registry (register, overwrite, function-name key); built-in metric functions across column types; is_text helper; _build_column_metrics (alias correctness, count_null derivation, summary merge, empty DataFrame, None-returning metrics). Existing integration tests already cover the refactored aggregation path end-to-end.

Documentation and Demos

• added/updated docs
• added/updated agent skills
• added/updated demos

New sections in data_profiling.mdx guide and profiler.mdx reference showing how to register custom metrics; dqx-profile-and-generate/SKILL.md updated with the extension point.

🤖 Generated with Claude Code

[mwojtyczka]

Reviewed the profiler column-metrics registry refactor. The extension-point design is reasonable, but the wiring has a blocking crash plus a few correctness regressions in _build_column_metrics — details inline.

[mwojtyczka]

Follow-up review of the head commit. The six earlier threads are all addressed — resolved them. A few new points on the extension point and conventions:

[IvannKurchenko]

Hello, @mwojtyczka! Thanks for a review. The previous feedback has been addressed. Would it be possible to have another round?

[mwojtyczka]

A few findings from a follow-up review pass (test-coverage and design-depth). None are correctness blockers.

[mwojtyczka]

Test coverage — this assertion can't detect the regression its name implies. Both test_empty_count_returns_column_for_text_types (line 28) and test_empty_count_returns_literal_zero_for_non_text_types (here) assert only empty_count(...) is not None. Since empty_count returns a Column in both branches (F.count_if(...) for text, F.lit(0) for non-text), is not None is trivially true either way — so if the non-text branch were changed from F.lit(0) to F.count_if(...), this test would still pass. The "returns literal zero for non-text" backward-compat guarantee the PR calls out is never actually exercised. Suggest evaluating the metric against a small Spark DataFrame and asserting the non-text result equals 0 (or at minimum asserting the two branches differ).

[mwojtyczka]

Altitude — one responsibility split across two methods. _build_column_metrics receives the whole summary_stats dict but only reads its own entry (summary_stats.get(field.name, {})), while the matching write summary_stats[field.name] = metrics lives in the caller _profile (line 445). A reader must hold both methods in mind to follow how summary_stats[field.name] evolves. Passing only the per-field entry in (e.g. field_summary_stats=summary_stats.get(field.name, {})) and letting _profile own the single assignment of the returned dict would make _build_column_metrics a pure function of its inputs.

[mwojtyczka]

Please add a test that verifies the custom metrics flow — register a new custom metric via register_profile_column_metric(...), run the profiler, and assert the custom metric's computed value appears in the returned summary_stats. I verified it live env and it seems to work fine but we need a test

[mwojtyczka]

Thanks for the fixes. The previous issues are resolved now. Left some additional small comments - mainly missing test coverage

[mwojtyczka]

count_null can raise TypeError from a user-registered metric. total_count - metrics.get("count_non_null", 0) subtracts a value taken straight from the registry aggregation. register_profile_column_metric explicitly supports overwriting keys (it logs "Overwriting profile column metric registered as ..."), so a user metric registered under count_non_null whose aggregation evaluates to SQL NULL on some column (e.g. F.max(...) over an all-null/empty column_df) lands None in metrics via asDict(), and int - None raises, aborting profiling. Consider coalescing the value to 0 before the subtraction.

[mwojtyczka]

Altitude: count/count_null are hardcoded special cases coupled to the literal key count_non_null, sitting outside the very registry this PR introduces. Overwriting the built-in count_non_null metric (a supported, warning-logged registry operation) silently makes count_null = total_count - 0 = total_count. Every downstream builder reads these with .get(key, 0) (e.g. make_null_or_empty_profile, the make_min_max_profile null-guard), so they then compute wrong null ratios and emit/suppress the wrong is_not_null/min_max rules with no error. The derived counts should either be first-class registry metrics or be protected from redefinition, rather than trusting a string key the extension point can overwrite.

[mwojtyczka]

Metric values reaching asDict() can be None, and consumers assume numbers. Once a registered metric's aggregation yields SQL NULL, field_aggregation_row.asDict() puts None into metrics. Consumers do arithmetic on these keys with only a missing-key guard — e.g. make_null_or_empty_profile (profile_builder.py:190) computes empty_count / total_count, so a metric registered under empty_count that evaluates to NULL gives None / total_count -> TypeError. .get(key, 0) guards absent keys but not present-but-None values, which the new registry makes reachable. Consider dropping/coalescing NULL aggregation results here before they enter the metrics dict.

[mwojtyczka]

Test mutates the process-global PROFILE_COLUMN_METRIC_REGISTRY, and the profiler runs every registered metric against every column. Registration here (and in the unit tests) is global with only a try/finally cleanup. Under pytest-xdist, or if a prior test errors outside its finally, a leaked registration is executed for all columns in unrelated profiling tests, altering their summary_stats and causing order-dependent failures. Consider a fixture that snapshots and restores the registry around each test rather than relying on manual pop().

[mwojtyczka]

Assertion on str(Column) tests an implementation detail. str(empty_count(field, "col")) == str(F.lit(0)) depends on the Column.__repr__/__str__ format, which is not a stable API and varies across Spark / Spark Connect versions. This can pass vacuously or break spuriously on a Spark upgrade without any behavior change, and it never actually exercises the count_if semantics. Prefer asserting on the evaluated aggregation result against a small DataFrame (that would be an integration test) or on the returned type.

[mwojtyczka]

Doc example registers a metric as an import-time side effect, with no opt-in or teardown note. The @register_profile_column_metric("percentile_10") decorator mutates the global PROFILE_COLUMN_METRIC_REGISTRY the moment this module/cell is imported, so simply importing or pasting the example permanently adds percentile_10 to every subsequent profile() call in the process (extra per-column aggregation, altered summary_stats). Worth a sentence noting this global, persistent effect and how to unregister, so users who copy it to "try it out" aren't surprised.