Skip to content

Uniqueness validation

Procedure and reference for SHACL-native global uniqueness (dqs:uniquenessConstraint / dqs:unique). For the mode overview and industrial example, see Uniqueness mode.

Prerequisites

  • SHACL/RuleSet published with dqs:uniquenessConstraint (or dqs:unique) on an indexed property — see Rule sources.
  • Validation infrastructure deployed (deploy_validation_infrastructure()); deploy creates dq-{view}-uniqueness when constraints are present — see Deploy → Scheduled uniqueness workflows.
  • Records stream configured if you consume DataQualityValidationRecord output — see Records output reference.

How uniqueness works

  • Uniqueness is declared in SHACL using dqs:uniquenessConstraint (or dqs:unique).
  • Deploy creates a dedicated scheduled workflow: dq-{view}-uniqueness.
  • The cron trigger omits job_run_id; runtime mints job_{view}_uniqueness_{epoch} for that scheduled execution. Explicit payload job_run_id is still honored. Uniqueness does not reuse the instance sync-cursor job id.
  • Runtime executes aggregate-first duplicate detection, then expands duplicates to instance-level results.
  • Results are written to DataQualityValidationRecord with dimensions set from dq:dimension (default "Uniqueness").
  • Duplicate failure writes are deduplicated: if a newer/equal prior failure already exists for the same focusNode, a new failure record is not written until the instance changes (lastUpdatedTime).

Hard requirement: property must be indexed

Uniqueness constraints are only supported on indexed properties in Data Modeling.

Reference: CDF Data Modeling - Indexes

What to expect in each run

Uniqueness emits typed records so apps and users can distinguish normal findings from overflow conditions.

From a user perspective:

  • Normal duplicate findings are emitted as validationType = "single_violation" when runtime expands duplicate groups to instance-level records.
  • Single-violation expansion is bounded by a run-level guardrail: up to 100 duplicate-value groups are expanded per run.
  • Grouped summary records are emitted as validationType = "group_violation" when guardrails or pass conditions apply:
  • groupViolationType = "global_overflow" when more than 100 different values have duplicates in the same run.
  • groupViolationType = "value_overflow" when one duplicate value has more than 100 violating instances.
  • groupViolationType = "pass" when no duplicates are found.
  • If you receive an overflow group record, treat it as "more duplicates remain" and rerun uniqueness after cleanup.

SHACL declaration (minimal)

@prefix sh:  <http://www.w3.org/ns/shacl#> .
@prefix dqs: <http://purl.org/cognite/dqs#> .
@prefix sp:  <http://purl.org/cognite/my_space/WorkOrder/> .

sp:WorkOrderUniqueness
    a sh:NodeShape ;
    sh:targetClass sp:WorkOrder ;
    dqs:uniquenessConstraint [
        dqs:property "workOrderNumber" ;
        dqs:ruleId "WO_Number_Unique" ;
    ] .

Rules:

  • dqs:property is required.
  • Optional dqs:ruleId on the uniqueness node is written as top-level ruleId on uniqueness records (filterable). Named uniqueness nodes use the node local name when dqs:ruleId is omitted. Blank nodes fall back to unique_{property}.
  • Grouped uniqueness externalId is still per property (dq_{ruleSetId}_group_{property}_...), not ruleId. Two uniqueness constraints on the same property share one grouped row on a mutable stream even when ruleId differs. Filter grouped rows by ruleId on an immutable stream, or keep one uniqueness rule per property.
  • The property must be indexed.
  • Grouped null values are ignored by the uniqueness executor.
  • Optional dq:dimension "Uniqueness" on the NodeShape (canonical dq: namespace) is copied onto uniqueness records as dimensions for UI filtering. Omit it and records still default to dimension Uniqueness.

Scheduling

Uniqueness runs on a dedicated schedule, not on every sync run.

  • Default schedule: 0 6 * * * (unless overridden)
  • Override in view YAML:
uniqueness_cron: "0 */4 * * *"   # every 4 hours
# uniqueness_cron: null           # disable uniqueness schedule

Output behavior

For a duplicate value with count N, runtime creates N failing instance-level results (one per focus node), unless suppressed by dedupe.

Dedupe rule for failures:

  • If latest existing failed record for (ruleSetId, ruleSetVersion, focusNode) has record.lastUpdatedTime >= instance.lastUpdatedTime, a new failure record is suppressed.
  • If the instance changes later, a new failure record is written again.

This prevents repeated failure spam for unchanged data.

How users detect "more duplicates remain"

From validation records:

  • Prefer typed signals over free-text parsing:
  • groupViolationType = "global_overflow" means run-level overflow.
  • groupViolationType = "value_overflow" means per-value overflow.
  • resultMessage can still include human-readable overflow details, but automation should rely on typed fields.
  • Fix known duplicates and rerun uniqueness until no overflow records appear and no new violations are reported.

Grouped uniqueness records (heuristic)

Grouped records are emitted by specific runtime guardrail conditions:

  • validationType = "group_violation" + groupViolationType = "global_overflow"
  • Trigger: more than 100 different values have duplicates in the same run.
  • Meaning: runtime emits a run-level overflow summary instead of expanding all duplicate-value groups.
  • validationType = "group_violation" + groupViolationType = "value_overflow"
  • Trigger: a specific duplicate value has more than 100 violating instances.
  • Meaning: runtime emits a per-value overflow summary for that value instead of writing all violating focus nodes.
  • validationType = "group_violation" + groupViolationType = "pass"
  • Trigger: no duplicate-value buckets are found (count <= 1 across returned buckets).
  • Meaning: grouped pass summary that confirms no duplicates were detected in that run.

These are grouped summaries and may omit focusNode / focusNodeInstance. They are not reverse-navigable from instance views — query them with the Records API instead (Reverse navigation). The grouped record externalId is keyed by property, not ruleId. Use groupViolationType as the primary machine-readable signal for investigation and rerun workflows.

Best practices

  • Keep uniqueness semantics in SHACL/RuleSet, not in ad-hoc scripts.
  • Build automations around validationType and groupViolationType, not free-text messages.
  • Treat overflow records as actionable backlog signals and rerun after cleanup.

Troubleshooting

  • Confirm SHACL contains dqs:uniquenessConstraint for the expected target class/property.
  • Confirm target property is indexed.
  • Confirm uniqueness workflow exists and has an active trigger.
  • dq:dimension "Uniqueness" is for record filtering only. It is not required for the constraint to run. Use the canonical namespace @prefix dq: <http://purl.org/cognite/cdf_dq/DataQuality/> .
  • Scheduled uniqueness omits job_run_id on the cron trigger and mints job_{view}_uniqueness_{epoch} for that run. A successful run logs job_run_id on the uniqueness banner and an [aggregate_uniqueness] outcome ... line with returnedBuckets, violatingBuckets, recordsPosted, recordsSuppressed, and truncated.
  • If the investigation script finds duplicates but Records are empty, inspect the function return JSON (not only the short log banner) for MISSING_JOB_RUN_ID, recordsPosted, and nested globalRuleResults[].result. A uniqueness error now fails the CDF function call.
  • If recordsPosted is low but duplicates still exist, check recordsSuppressed (dedupe may be working as intended).
  • If grouped records include groupViolationType = "global_overflow" or groupViolationType = "value_overflow", fix current duplicates and rerun to continue through backlog.

Pro tip: list all duplicates with a Python script

If you see grouped overflow records (groupViolationType = "global_overflow" or "value_overflow"), you can run a one-off script to inspect all current duplicate values directly.

What this script does:

  1. Connects with your config.toml credentials.
  2. Runs one aggregate query grouped by the uniqueness property.
  3. Keeps only values with count > 1 (duplicate values).
  4. For each duplicate value, runs instances.list to fetch impacted instances.
  5. Prints duplicate groups ordered by duplicate count (largest first).
from __future__ import annotations

from pathlib import Path

from cognite.client import data_modeling as dm
from cognite.client.data_classes.data_modeling import ViewId

from cognite_data_quality import load_cognite_client_from_toml


def main() -> None:
    # --- Change these values for your setup ---
    toml_path = Path("config.toml")  # quickstart format (client_id / client_secret)
    data_product_id = "enterprise-process-industry"
    data_product_version = "1.0.0"  # published version from the Data Products UI
    view_space = "sp_enterprise_process_industry"
    view_external_id = "YourOrgMaintenanceOrder"
    view_version = "v1"
    uniqueness_property = "name"
    # Optional filter by instance spaces from your data contract view definition:
    instance_spaces: list[str] = ["test_data_pumps"]
    # ------------------------------------------

    client = load_cognite_client_from_toml(toml_path)

    print(f"Data Product: {data_product_id}@{data_product_version}")

    view_id = ViewId(view_space, view_external_id, view_version)
    aggregate_filter = (
        {"in": {"property": ["node", "space"], "values": instance_spaces}}
        if instance_spaces
        else None
    )

    # 1) Aggregate duplicates by property value
    buckets = list(
        client.data_modeling.instances.aggregate(
            view=view_id,
            aggregates=[dm.aggregations.Count("externalId")],
            group_by=[uniqueness_property],
            instance_type="node",
            filter=aggregate_filter,
            limit=1000,  # CDF API max per request
        )
        or []
    )

    # 2) Keep only duplicate values (count > 1)
    duplicate_groups: list[tuple[str, int]] = []
    for bucket in buckets:
        value = (bucket.group or {}).get(uniqueness_property)
        count_obj = (bucket.aggregates or [None])[0]
        count = int(getattr(count_obj, "value", 0) or 0)
        if value is not None and count > 1:
            duplicate_groups.append((str(value), count))

    # 3) Order groups by duplicate size (largest first)
    duplicate_groups.sort(key=lambda item: item[1], reverse=True)
    print(f"Duplicate value groups found: {len(duplicate_groups)}")

    property_path = [view_space, f"{view_external_id}/{view_version}", uniqueness_property]

    for duplicate_value, duplicate_count in duplicate_groups:
        print("\n" + "=" * 100)
        print(f"Value: {duplicate_value!r}  |  Count: {duplicate_count}")

        value_filter: dict[str, object] = {
            "equals": {"property": property_path, "value": duplicate_value}
        }
        detail_filter = {"and": [aggregate_filter, value_filter]} if aggregate_filter else value_filter

        # 4) List impacted instances for this duplicate value
        rows = list(
            client.data_modeling.instances.list(
                sources=[view_id],
                instance_type="node",
                filter=detail_filter,
                limit=1000,  # CDF API max per request
            )
            or []
        )

        for row in rows:
            print(f"- {row.space}/{row.external_id} (last_updated_time={row.last_updated_time})")


if __name__ == "__main__":
    main()

Use this as an investigation tool when users need full duplicate visibility before cleanup planning.

Note:

  • The examples above use limit=1000, which is the API maximum per request for these calls.
  • If a duplicate group contains more than 1000 instances, fetch additional pages (for example by partitioning filters) to retrieve the full set.

Previous section

Next section