Records output reference

Field reference for validation and inference records written to the CDF Records API. For stream mutability decisions and operator guidance, see Records streams (mutable vs immutable). For query patterns in notebooks, see the example notebook.

Record types

Container	Purpose
`DataQualityValidationRecord`	One record per focus node per run — pass/fail, violation details, severity
`RuleEngineResult`	One record per inference result — derived state from SHACL-AF `sh:rule` blocks

Both containers live in the same configurable space (default dataQuality) and stream.

Consumption surfaces (0.5)

On 0.4.x, dashboards and notebooks query the Records API with container + stream property paths (examples below).

On 0.5.x, the supported Flows contract is view + stream (see Record views (Flows) below). Opt-in instance → records navigation is a separate customer model change — Reverse navigation from instance views.

Structured SHACL sh:name on validation results is a separate 0.5 concern (constraint display names). Reverse navigation is instance ↔ records; it does not replace per-rule names. failedConstraints semantics stay as documented in the table below.

Grouped uniqueness record mode

To support high-volume grouped violations (uniqueness overflow and green pass markers), DataQualityValidationRecord includes an explicit record mode:

validationType = "single_violation" for existing per-instance records
validationType = "group_violation" for grouped summary records

Grouped records also include:

groupViolationType subtype:
global_overflow
value_overflow
pass

For group_violation records, focusNode and focusNodeInstance may be intentionally omitted. Those records are not reverse-navigable from instance views (same for RAW rows). Query them with the Records API instead — see Reverse navigation. Consumers (apps/agents) should branch on validationType and groupViolationType rather than parsing free text.

Default configuration

Records are written to:

Space: dataQuality (configurable via records.space / records_space in settings)
Container: DataQualityValidationRecord (configurable via records.container in settings)
Stream: records.stream_id (e.g. dq_validation_stream, test_stream_1)
Stream template (creation only): records.stream_type — CDF template name BasicArchive (default), BasicLiveData, or ImmutableTestStream
Rule engine (optional): records.rule_engine_container / records.rule_engine_stream_id to route inference to a separate stream

To post results when running validation locally, set post_to_records=True and provide a stream_id in RecordsConfig when calling run_validation.

Stream mutability (mutable vs immutable)

Operator decision guide: Records streams (mutable vs immutable).

A stream's mutability is fixed when it is created and cannot be changed. The configured stream therefore dictates how records are written — the code adapts at runtime by auto-detecting the stream's type (via GET /streams/{id}) and choosing the write endpoint. Successful lookups are cached in a module-global cache for the function process lifetime (warm invocations may reuse it):

Stream type	Template (`records.stream_type`)	Write endpoint	Behavior
Immutable	`BasicArchive`	`POST /streams/{id}/records` (ingest)	Every run is kept. Full history / audit. Higher record caps. Filter/aggregate queries require a `lastUpdatedTime` range.
Mutable	`BasicLiveData`	`POST /streams/{id}/records/upsert`	One record per stable key (shape varies — instance, edge, uniqueness group, timeseries, schema issue). De-duplicated statistics, no `lastUpdatedTime` window needed. Smaller cap (~5M records).

Records use a stable externalId (for example dq_{ruleSetId}_{instanceSpace}_{instanceExternalId}) that no longer embeds jobRunId. All information previously encoded in the ID remains available as queryable properties (ruleSetId, focusNodeInstance, jobRunId, plus schema-issue details in validationReport). Consequences:

Mutable stream: re-validating the same focus upserts its record for that stable-key dimension (instance+ruleset, edge relation, group summary, etc.). Best for current-state statistics and high ruleset-update churn.
Immutable stream: the stable externalId is no longer unique per run — a single externalId can match multiple records (one per run), distinguished by the jobRunId property and createdTime. Read paths that need a specific run must filter/sort by jobRunId and/or lastUpdatedTime, not assume ID uniqueness.

records.stream_type (CDF template name in settings, stored as records_stream_type in DMS) only affects stream creation; at runtime the actual stream type is always auto-detected.

Stream type auto-detection. The handler calls GET /streams/{id} (alpha API). On any error (missing streamsAcl:READ, stream not found, network/timeout, unknown type field) it fail-safes to immutable ingest. That is correct for an immutable stream, but on a mutable stream a duplicate ingest is rejected (409/422). Transient failures are not cached and retry on the next call; successful lookups are cached for the process lifetime. Grant streamsAcl:READ before pointing production at a mutable stream. See Deploy → Records stream type and permissions.

Stable key uniqueness and instance space

The instance record key is dq_{ruleSetId}_{instanceSpace}_{instanceExternalId} — the instance space is part of the key. This means the same externalId reused in different instance spaces maps to distinct record keys, so a mutable upsert never overwrites across spaces. RAW rows (which have no instance space) keep the space-less form dq_{ruleSetId}_{rowKey}; RAW row keys are unique within a table. RAW validation uses the same stream mutability auto-detection as instance validation — choose BasicArchive or BasicLiveData when creating the stream. (DMS space IDs are short and use the external-ID-safe charset, so the raw space is embedded directly rather than hashed.)

Migration: mixed external-ID schemes

External IDs changed from the previous run-scoped form (dq_{ruleSetId}_{instanceExternalId}_{jobRunId}) to the stable, space-qualified form (dq_{ruleSetId}_{instanceSpace}_{instanceExternalId}). After upgrading, an existing stream will hold both schemes until the old run-scoped records age out (immutable retention) or are removed. Statistics that assume a single scheme should account for the mix during the transition, or start from a fresh stream.

Dashboard note: the Streamlit validation dashboard currently assumes immutable streams (it uses lastUpdatedTime windows and treats each run as a distinct record). If your dashboard queries records that way, keep the validation stream on BasicArchive rather than BasicLiveData. Mutable streams upsert by stable externalId and are intended for current-state KPIs — see Records streams. Writing the same validation results to both stream types at once is not supported.

DataQualityValidationRecord fields

Each record in the stream has a properties object keyed by space and container. The data quality fields live under properties[records_space][records_container] (e.g. dataQuality / DataQualityValidationRecord). Main properties:

Property	Type	Description
`ruleSetId`	string	Rule set identifier — the primary focus RuleSet `externalId` for the view after DP load (not necessarily `quality.rules[0]`). When several RuleSets still match the view class, exact focus (`focus` target classes equal the view URI) wins over multi-class matches and over `quality.rules` list order; falls back to view `externalId` when nothing matches. See Building a quality app — Engine focus binding
`ruleSetVersion`	string	Rule set version from the primary `ruleset_references` entry for that `ruleSetId`
`jobRunId`	string	Unique identifier for the validation job run (property only — not part of `externalId`)
`validationType`	string	`"single_violation"` (default per-focus) or `"group_violation"` (grouped uniqueness summaries)
`groupViolationType`	string	Present on grouped uniqueness only: `"global_overflow"`, `"value_overflow"`, or `"pass"`
`passedValidation`	boolean	`true` if no Violation-severity findings, `false` otherwise
`resultSeverity`	list of string	Severities from violations: `Violation`, `Warning`, `Info` (empty on pass)
`failedConstraints`	list of string	Failed constraint identifiers (e.g. `MinCountConstraintComponent::name`, `dqs:edge_existence::workItems`, `dqs:UniquenessConstraint::workOrderNumber`). Machine tokens — not overwritten with `sh:name`
`failedConstraintNames`	list of string	Deduped human-readable names for failed constraints (`sh:name` from the shapes graph, or a typed token for uniqueness / edge existence). Empty when validation passed. Not a parallel array to `failedConstraints`
`ruleId`	string	Unique identifier for the rule that produced this record. Written for uniqueness / aggregate results (`dqs:ruleId` when set in SHACL, otherwise the named uniqueness node or `unique_{property}`). Omitted on other validation paths. Not a display label — use `failedConstraintNames` / `shaclName` for names. Grouped uniqueness `externalId` is still per property, so two uniqueness rules on the same property share one grouped key even when `ruleId` differs
`focusNode`	string	URI of the validated instance (focus node). Omitted on `group_violation` records
`focusNodeInstance`	object	Direct relation to the node instance (`space`, `externalId`). Omitted for RAW rows, `group_violation`, and edge-view SHACL (focus is the edge URI in `focusNode`; edges are not node instances)
`sourceShape`	list of string	Source shape URIs from SHACL (when available)
`dimensions`	list of string	Quality-dimension annotations from SHACL (`dq:dimension`); see Dimensions
`dataDomainExternalId`	string	Optional data domain / view / DataProduct tag (written by some builders; not every handler path)
`validationReport`	object	Full report with `violations`, `violationCount`, `summary` (plus type-specific detail). Each SHACL violation may include `shaclName` when the shapes graph defines `sh:name`

Record-level fields (outside properties) include externalId, createdTime, and lastUpdatedTime.

Constraint display names (`sh:name`)

SHACL sh:name is looked up on the shapes graph (the same parent-walk used for dq:dimension). The validation report's sourceConstraint blank node is not a Turtle node and is not used as a key.

Resolution order:

sh:name on the failing SPARQL or property constraint
else sh:name on the enclosing sh:property shape
else sh:name on the source NodeShape

If no sh:name is present, shaclName is omitted and failedConstraintNames does not invent a label from resultMessage.

Same-path / multi SPARQL: two (or more) sh:sparql constraints on the same sh:path, or multiple pathless SPARQL on one NodeShape, share one failedConstraints machine token (SPARQLConstraintComponent::{path} or the bare component). That is intentional — do not write one Records row per failed rule. Identity for those rules is failedConstraintNames / validationReport.violations[].shaclName. The writer matches a violation to a constraint by comparing the report resultMessage (with trailing [dimension] [severity] suffixes stripped) to indexed sh:message literals. When two named SPARQL constraints cannot be matched that way, shaclName is omitted (no NodeShape fallback, no scrape of free-text messages as names). Author every SPARQL constraint with a unique sh:name and a stable sh:message (prefer dq:ruleId as well).

Non-SHACL writers set a typed display name on the same fields:

Path	`shaclName` / `failedConstraintNames`
Edge existence	`dqs:edge_existence::{relationProperty}`
Uniqueness (single or grouped)	`dqs:UniquenessConstraint::{property}`
Schema issue / engine failure	omitted (empty list)

Filter or aggregate display names on failedConstraintNames. Per-violation names live on validationReport.violations[].shaclName. Keep grouping by machine token on failedConstraints for non-SPARQL components; for same-path SPARQL, dashboards and Compliance UIs must use failedConstraintNames / per-violation shaclName.

Record views (Flows)

deploy_validation_infrastructure() ensures two record views in the records space (default dataQuality), version v1:

View `externalId`	Maps container	`streamId`
`DataQualityValidationRecord`	`DataQualityValidationRecord`	`records.stream_id`
`RuleEngineResult`	`RuleEngineResult`	`records.rule_engine_stream_id` or `records.stream_id`

Both views are applied in one request. They map every container property, including failedConstraintNames and focusNodeInstance. Record views are not added to a data model (the platform does not accept record views in data models yet). They are not GraphQL or instances.query sources. Flows binds view + stream.

If the Records stream does not exist yet, view apply is skipped with a warning and the rest of infrastructure deploy continues. Create the stream (CDF Console/API, or ensure_records_infrastructure()), then re-run deploy.

Opt-in reverse navigation (multi_reverse_direct_relation through focusNodeInstance on your instance views): copy-paste YAML and non-navigable cases in Reverse navigation from instance views.

Stable `externalId` patterns

Kind	Pattern
Instance / uniqueness single	`dq_{ruleSetId}_{instanceSpace}_{externalId}` (space omitted if empty)
Edge existence	`dq_{ruleSetId}_edge_{relationProperty}_{instanceSpace}_{externalId}`
Uniqueness group	`dq_{ruleSetId}_group_{property}_global` / `..._pass` / `..._{duplicateValue}` (property key, not `ruleId`)
Schema issue	`schema_{ruleSetId}_{issueType}_{viewId}_{property}_{idx}`
Timeseries	`dq_{ruleSetId}_{tsSpace}_{tsId}` (backfill may append a time-window suffix)

Edge existence records

Edge existence uses the same container with validationType = "single_violation". Focus is always the node (never the edge URI). Failures set failedConstraints to dqs:edge_existence::{relationProperty} and report sourceConstraintComponent: "dqs:edge_existence". PASS and FAIL are both written. See Edge existence.

Query examples (Records API)

The Records API is in alpha. Use the CDF alpha header for all requests:

client.config.headers["cdf-version"] = "alpha"

Base URL pattern: /api/v1/projects/{project}/streams/{stream_id}/records.

Filter records

Endpoint: POST .../streams/{stream_id}/records/filter

Request body:

filter (optional): Filter on container properties using equals, in, etc.
lastUpdatedTime: Time range, e.g. {"gt": start_ms} or {"gte": start_ms, "lte": end_ms} (milliseconds since epoch). Required on an immutable stream. Omitting it returns 400. Some projects reject a range wider than about 7 days (interval-too-big); split the window and combine the pages. Keep the same range on every cursor page.
limit (optional): Page size. Follow nextCursor until it is absent. 1000 is not the whole population.
cursor (optional): For pagination. Do not change lastUpdatedTime while a cursor is in use.

Property paths in filters are arrays: [space, container, propertyName], e.g. ["dataQuality", "DataQualityValidationRecord", "ruleSetId"].

Example: fetch validation records for a rule set (last 7 days)

from cognite.client import CogniteClient
from datetime import datetime, timedelta

client = CogniteClient()
client.config.headers["cdf-version"] = "alpha"

stream_id = "dq_validation_stream"  # or your settings["records"]["stream_id"]
space = "dataQuality"
container = "DataQualityValidationRecord"
rule_set_id = "EquipmentSHACLv1"

seven_days_ago_ms = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)

response = client.post(
    f"/api/v1/projects/{client.config.project}/streams/{stream_id}/records/filter",
    json={
        "lastUpdatedTime": {"gt": seven_days_ago_ms},
        "limit": 1000,
        "filter": {
            "equals": {
                "property": [space, container, "ruleSetId"],
                "value": rule_set_id,
            }
        },
    },
)
data = response.json()
records = data.get("items", [])

for record in records:
    props = record.get("properties", {}).get(space, {}).get(container, {})
    passed = props.get("passedValidation")
    focus = props.get("focusNode")
    failed = props.get("failedConstraints", [])
    print(f"  {record.get('externalId')}: passed={passed}, focus={focus}, failed={len(failed)}")

Example: filter by pass/fail

passedValidation is false when any stored violation exists, including schema checks and rules the screen is not listing. It is a diagnostic filter, not the compliance card total. Listed-rule scores are in Building an app on the data quality engine. On an immutable stream the time range stays on this request too.

# Stored violations in the window — not the listed-rule fail count
response = client.post(
    f"/api/v1/projects/{client.config.project}/streams/{stream_id}/records/filter",
    json={
        "lastUpdatedTime": {"gt": seven_days_ago_ms},
        "filter": {
            "equals": {
                "property": [space, container, "passedValidation"],
                "value": False,
            }
        },
        "limit": 500,
    },
)

Aggregate records

Endpoint: POST .../streams/{stream_id}/records/aggregate

Use this to get counts, unique values, or other aggregates without loading all records.

Request body:

filter (optional): Same shape as filter (e.g. by ruleSetId, jobRunId). Do not put a listed-rule or inside an aggregate filters map; put it on this request filter. The API rejects the other shape and the caller sees a zero window.
lastUpdatedTime: Time range in milliseconds. Required on an immutable stream, including aggregates. Use the same range you would send to filter.
aggregates: Map of aggregate names to definitions, e.g.:
"totalCount": {"count": {}}
"uniqueJobs": {"uniqueValues": {"property": [space, container, "jobRunId"]}}
"passedBreakdown": {"uniqueValues": {"property": [space, container, "passedValidation"]}}
"severityBreakdown": {"uniqueValues": {"property": [space, container, "resultSeverity"]}}

Example: aggregate stats for a rule set

payload = {
    "lastUpdatedTime": {"gt": seven_days_ago_ms},
    "filter": {
        "equals": {"property": [space, container, "ruleSetId"], "value": rule_set_id}
    },
    "aggregates": {
        "totalCount": {"count": {}},
        "jobRuns": {"uniqueValues": {"property": [space, container, "jobRunId"]}},
        "passedBreakdown": {"uniqueValues": {"property": [space, container, "passedValidation"]}},
    },
}
response = client.post(
    f"/api/v1/projects/{client.config.project}/streams/{stream_id}/records/aggregate",
    json=payload,
)
data = response.json()

total = data.get("aggregates", {}).get("totalCount", {}).get("count", 0)
job_buckets = data.get("aggregates", {}).get("jobRuns", {}).get("uniqueValueBuckets", [])
unique_jobs = len(job_buckets)
# passedBreakdown.uniqueValueBuckets: [{"value": true, "count": n}, {"value": false, "count": m}]

Example: list distinct rule set IDs in the stream

payload = {
    "lastUpdatedTime": {"gt": seven_days_ago_ms},
    "aggregates": {
        "ruleSetIds": {"uniqueValues": {"property": [space, container, "ruleSetId"]}}
    },
}
response = client.post(
    f"/api/v1/projects/{client.config.project}/streams/{stream_id}/records/aggregate",
    json=payload,
)
data = response.json()
buckets = data.get("aggregates", {}).get("ruleSetIds", {}).get("uniqueValueBuckets", [])
rule_set_ids = [b["value"] for b in buckets if b.get("value")]

Restoring the client default headers

After querying Records, restore the client headers if you need to avoid sending cdf-version: alpha to other APIs:

original_headers = client.config.headers.copy()
client.config.headers["cdf-version"] = "alpha"
try:
    # ... Records API calls ...
finally:
    client.config.headers = original_headers

RuleEngineResult fields

When a SHACL file contains sh:rule / sh:SPARQLRule shapes (SHACL-AF), the engine runs inference automatically alongside quality validation. Each CONSTRUCT triple typed dqs:RuleEngineResult is written as a separate record in the RuleEngineResult container.

Property	Type	Description
`ruleSetId`	string	RuleSet external ID the rule came from
`ruleSetVersion`	string	RuleSet version
`ruleId`	string	Value of `dqs:ruleId` on the SHACL rule
`runId`	string	Shared job run ID (same as `DataQualityValidationRecord.jobRunId`)
`resultType`	string	Always `"Inference"` for rule engine results
`focusNode`	string	URI of the focus node
`focusNodeInstance`	DirectRelation	Hard link to the DMS instance (space, externalId)
`resultValue`	string	Optional scalar value (e.g. `"Critical"`, `"Overdue"`)
`resultPayload`	JSON	Optional free-form JSON body
`causedBy`	DirectRelation list	Lineage — upstream `RuleEngineResult` records this result depends on
`dataDomainExternalId`	string	Optional domain tag
`producedAt`	Timestamp	Materialisation time

Querying RuleEngineResult records

client.config.headers["cdf-version"] = "alpha"

rule_engine_stream_id = "dq_validation_stream"  # same stream as DQ records by default
space = "dataQuality"
container = "RuleEngineResult"

# Fetch all "MO_Criticality" inferences produced today
response = client.post(
    f"/api/v1/projects/{client.config.project}/streams/{rule_engine_stream_id}/records/filter",
    json={
        "filter": {
            "equals": {
                "property": [space, container, "ruleId"],
                "value": "MO_Criticality",
            }
        },
        "limit": 1000,
    },
)
for record in response.json().get("items", []):
    props = record["properties"][space][container]
    print(f"{props['focusNode']}: {props.get('resultValue')}")

Incremental reads for rule chaining

For downstream rule stages, use the rule_engine_result_sync function handler.

It supports both Records sync mode and latest-changed filter mode.
It persists per-listener checkpoint state (cursor + watermark).
It supports first-class rule-scoped filters (rule_ids, rule_set_ids, rule_set_versions, focus_nodes, focus_node_instances) in both modes.
It supports sync pagination beyond 1,000 records per run via page_limit + max_records.

For end-to-end chained pipeline usage, see Chained conditional logic.

Writing inference rules (SHACL-AF)

See Rule sources for the full authoring guide.

Best practices

Use typed fields (validationType, groupViolationType, ruleId) for automation instead of parsing message text.
Keep dashboard queries time-bounded with lastUpdatedTime.
Separate operational and exploratory queries by stream and filter scope.

Troubleshooting

Empty query results: verify stream/space/container and alpha header usage.
Missing grouped uniqueness signals: confirm uniqueness workflow is scheduled and active.
Missing inference lineage: verify downstream rules populate causedBy.

Dashboard

You can build your own dashboard using the filter and aggregate endpoints above. On an immutable stream every call sends lastUpdatedTime. Query by ruleSetId, jobRunId, failedConstraintNames, and failedConstraints. Do not treat passedValidation as the pass/fail total for the rules on screen. Count listed-rule hits in a separate request from the record total. The full contract is Building an app on the data quality engine. The RuleEngineResult container supports the same filter/aggregate API and can feed KPI dashboards directly.

Previous section

Rule sources

Next section

Reverse navigation from instance views