Fix duplicate-key row multiplication in dataset comparisons

[mattfaltyn]

Changes

• Prevent duplicate-key Cartesian fan-out in compare_datasets with lazy one-to-one pairing by default. Rows within each matching-key group receive a row_number() ordered by the string representation of compared values, with nulls first.
• Add raise_on_duplicate_keys=True to opt into eager uniqueness validation on both datasets. With standard null matching, rows containing null matching keys are excluded from uniqueness validation because they cannot match.
• Use the generated row number to distinguish present null-key rows from missing rows during default pairing.
• Document lexical numeric ordering and that pairing does not minimize reported changes.
• Explicitly reject streaming DataFrames on either side and document the batch-only restriction in the changelog.
• Restore repeated benchmark measurements for both lazy pairing and strict validation, including check construction in the timed section.

This prevents many-to-many joins from multiplying output rows and reporting false changes between identical datasets, while keeping the default comparison lazy.

Linked issues

Resolves #1487

Tests

• manually verified duplicate-key join cardinality and existing unique-key comparisons
• added unit tests for streaming rejection on either side in both modes
• added integration coverage for strict validation, lazy construction, null keys, duplicate pairing, and unmatched rows
• added end-to-end tests — no end-to-end surface changed
• updated performance benchmarks for both modes — workspace-scale benchmarks have not been run locally

Local validation:

• make fmt
• make lint
• make test — 2,484 passed
• Standalone PySpark 4.0.0 comparison checks — 27 passed using unchanged repository test bodies with local Spark/UTC fixtures; the Unity Catalog reference-table case was excluded
• UV_FROZEN=1 UV_BUILD_CONSTRAINT=.build-constraints.txt uv run --exact --all-extras --group docs pydoc-markdown — succeeded with three cross-reference warnings in unchanged aggregate helpers
• corepack yarn --cwd docs/dqx build — 150 documents indexed, production build succeeded

The repository's workspace-backed integration fixture could not initialize because DATABRICKS_HOST is unset. The local Spark checks do not replace Databricks Connect, Unity Catalog, or workspace-scale benchmark validation.

Documentation and Demos

• added/updated demos — no demo flow changed
• updated docstrings, reference documentation, and changelog
• added/updated agent skills — not applicable

[ghanse]

Left a comment here regarding design: #1487 (comment)

Let me know what you think.

[mattfaltyn]

Failing fast with an InvalidParameterError is a reasonable default. I think we should optionally support duplicate keys.

Thanks @ghanse. Implemented in 41d9bc9. Strict uniqueness remains the default; allow_duplicate_keys=True opts into row_number() pairing ordered by compared values. Added docs, tests, and benchmark coverage.

[ghanse]

Solid PR overall — it matches the design we landed on in #1487 (deterministic pairing for duplicate keys), with good test and doc coverage, and I didn't spot any correctness bugs.

My one real design question is whether the default should raise at all: raising needs an eager action, whereas the row_number() pairing you added is already lazy and already fixes the fan-out — so it might make a better default, with the hard-fail as an opt-in. Not a hard blocker, but worth a discussion. The rest are minor.

[ghanse]

The bigger thing: this default path fires two eager Spark jobs (.isEmpty() on each side) on every comparison, even when there are no dupes — but compare_datasets is otherwise lazy like the rest of our checks. The catch is you can't raise without an action, so any "validate then raise" is inherently eager.

Worth flipping this around: the row_number() pairing you added below is already fully lazy and already prevents the fan-out that #1487 is about. If that's the default, the bug is fixed with zero extra jobs, and unique keys are unchanged (rn=1 on both sides, so joining on (pk, rn) is the same as joining on pk). If we still want to surface non-unique keys, we can do it lazily instead of raising:

duplicate_key_condition = F.count(F.lit(1)).over(Window.partitionBy(*pk_column_names)) > 1

and fold it into the status JSON like row_missing/changed. A hard raise-and-stop is still reasonable, but since it's inherently eager I'd make it the opt-in flag (raise_on_duplicate_keys=True) rather than the default. Curious what you think — happy to be talked out of it if the raise needs to be the default.

[mattfaltyn]

If that's the default, the bug is fixed with zero extra jobs

Agreed, thanks! Pairing is now lazy by default; raise_on_duplicate_keys=True opts into eager validation. Added regressions for laziness and null-key pairing.

[ghanse]

Good guard — the window/groupBy machinery wouldn't behave on streaming anyway, so a clear error beats a cryptic one. Just flagging it's technically a new restriction for any streaming caller, so worth a line in the changelog/release notes.

[mattfaltyn]

worth a line in the changelog/release notes.

Added the batch-only restriction to the breaking-change notes, with guidance for streaming callers. Thanks for flagging it!

[ghanse]

Minor: casting everything to string means numerics order lexically ("10" < "2"). Doesn't affect correctness — the diff runs on the real values and identical rows still cluster together — just noting the pairing order is a bit arbitrary for numbers. Fine to leave as-is.

[mattfaltyn]

Fine to leave as-is.

Thanks! Kept the ordering and documented lexical numeric sorting; comparisons still use the original values.

[ghanse]

pedantic(rounds=1, iterations=1) is a single measurement with no repeats, so this won't catch perf regressions anymore — it's basically a smoke test now. Was that intentional? If you want a real signal, keeping the default multi-round benchmark() on at least the strict case would help.

[mattfaltyn]

keeping the default multi-round benchmark() on at least the strict case would help.

Good catch! Restored multi-round benchmark() for both modes, including check construction so strict validation is measured.