From fe16cc78c09302429b59290175f86166936a1d37 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 16:53:36 +1100 Subject: [PATCH 1/7] fix(schemas): serve the column vocabulary The console kept its own copies of the primitives, use cases and override types, and they drifted from dfe-schemas. It offered retired use cases and had no map or vector. - GET /api/v1/schemas/vocabulary serves primitives, use cases and override types from the pinned dfe-schemas registry, with labels and the names each use case replaced. - GET /api/v1/system/defaults carries the retention maximum beside the value. - The external pipeline header table uses the header's @source expressions. It no longer shows @timestamp as the source for _timestamp. Labels and override options live in dfe-schemas types.yaml. Until the pin moves to a release that has them, labels use the names and no override types are offered. --- docs/data-plane/external-pipelines.md | 12 +- docs/data-plane/schema.md | 2 + docs/reference/api-route-map.md | 5 +- openapi/openapi.json | 305 +++++++++++++++++- scripts/route-plan.toml | 1 + src/dfe_engine/api/v1/schemas.py | 88 +++++ src/dfe_engine/api/v1/sources.py | 4 +- src/dfe_engine/api/v1/system.py | 2 + src/dfe_engine/source/external.py | 124 +++++-- src/dfe_engine/source/type_registry.py | 175 ++++++++++ tests/unit/test_api/test_schemas_router.py | 39 +++ tests/unit/test_api/test_system_defaults.py | 2 + .../test_source/test_external_pipeline.py | 54 +++- tests/unit/test_source/test_type_registry.py | 134 ++++++++ 14 files changed, 903 insertions(+), 44 deletions(-) diff --git a/docs/data-plane/external-pipelines.md b/docs/data-plane/external-pipelines.md index 75e9c013e..c503cfad5 100644 --- a/docs/data-plane/external-pipelines.md +++ b/docs/data-plane/external-pipelines.md @@ -105,13 +105,13 @@ Nothing makes your records carry the DFE common header. The loader fills in a mi | Field | Missing, the loader | Set, you get | |---|---|---| -| `_timestamp` | reads `timestamp`, then `@timestamp`, and takes the load time when neither is there | the event's own time, so time-range queries and retention follow when it happened | -| `_org_id` | lands the record in the source's org | the source's org or one `allowed_org_ids` lists, which is how a provider lands each customer in their own org | -| `_source` | takes it from the topic name, which is this source | this source or one `allowed_sources` lists, which picks the table | -| `_tags` | reads `tags`, then `meta`, then `metadata.tags`, and leaves it empty when none is there | tags a query or an alert can filter on without parsing the payload | -| `_timestamp_received` | takes the load time | when your pipeline first saw the event, so you can measure the delay before DFE | +| `_timestamp` | reads `timestamp`, and takes the load time when it is not there | the event's own time, so time-range queries and retention follow when it happened | +| `_org_id` | reads `_org_id`, and lands the record in the source's org when it is not there | the source's org or one `allowed_org_ids` lists, which is how a provider lands each customer in their own org | +| `_source` | reads `_source`, and takes the topic name, which is this source, when it is not there | this source or one `allowed_sources` lists, which picks the table | +| `_tags` | reads `_tags`, then `tags`, then `meta`, then `metadata.tags`, and leaves it empty when none is there | tags a query or an alert can filter on without parsing the payload | +| `_timestamp_received` | reads `_timestamp_received`, then `timestamp_received`, then `received_at`, and takes the load time when none is there | when your pipeline first saw the event, so you can measure the delay before DFE | -`_timestamp_load`, `_uuid`, `_json` and `_raw` are always written by the loader and ClickHouse, and a value you send isn't read. The columns come from the source's header profile in dfe-schemas (`common-header/`), and the default profile is `timeseries`. +`_timestamp_load`, `_uuid`, `_json` and `_raw` are always written by the loader and ClickHouse, and a value you send isn't read. The columns and the order the loader reads them in come from the source's header profile in dfe-schemas (`common-header/`), and the default profile is `timeseries`. A compliant record looks like this: diff --git a/docs/data-plane/schema.md b/docs/data-plane/schema.md index 0d3a0a996..2253cf873 100644 --- a/docs/data-plane/schema.md +++ b/docs/data-plane/schema.md @@ -373,6 +373,8 @@ catalogue: | **Enum** | `Enum8(...)`, `Enum16(...)` | | **Special** | `Dynamic`, `Variant(...)`, `AggregateFunction(...)`, `SimpleAggregateFunction(...)` | +`GET /api/v1/schemas/vocabulary` serves what a schema form offers, read from this registry: the primitives, the index use cases with their labels and the names they replaced, the attributes, and the override types with the primitive each refines. A type that takes parameters carries a `template` such as `Decimal({precision},{scale})`, so a form builds the override text without knowing the type. A label the registry does not declare reads off the entry's name. The retired `lowcardinality` attribute is not offered: a column declares `cardinality`. + --- ## Cardinality Axis diff --git a/docs/reference/api-route-map.md b/docs/reference/api-route-map.md index e0db7cf88..5b59d3946 100644 --- a/docs/reference/api-route-map.md +++ b/docs/reference/api-route-map.md @@ -10,11 +10,11 @@ One row per operation the engine serves today. **New** is where it goes, and its |---|---|---| | moved | 87 | A new path, method or both | | merged into | 9 | Folded into the operation **New** names | -| unchanged | 41 | Same method and path | +| unchanged | 42 | Same method and path | | frozen | 54 | Kept as it is | | decide | 129 | Not placed yet: see its question | | removed | 0 | Gone, with nothing in its place | -| total | 320 | | +| total | 321 | | ## Questions @@ -180,6 +180,7 @@ One row per operation the engine serves today. **New** is where it goes, and its | `DELETE /api/v1/schemas/definitions/{schema_path}/versions/{version}` | `DELETE /api/v1/schemas/{schema}/versions/{version}` | `schemas.versions.delete` | `dfe schemas versions delete` | `dfe_schemas_versions_delete` | moved | ui 2 | | | `POST /api/v1/schemas/elastic-converter` | `POST /api/v1/schemas/elastic-converter` | `schemas.elastic-converter` | `dfe schemas elastic-converter` | `dfe_schemas_elastic_converter` | unchanged | ui 2 | | | `POST /api/v1/schemas/import` | `POST /api/v1/schemas/import` | `schemas.import` | `dfe schemas import` | `dfe_schemas_import` | unchanged | 0 | | +| `GET /api/v1/schemas/vocabulary` | `GET /api/v1/schemas/vocabulary` | `schemas.vocabulary` | `dfe schemas vocabulary` | `dfe_schemas_vocabulary` | unchanged | 0 | | ## sources diff --git a/openapi/openapi.json b/openapi/openapi.json index 461fbb026..c821d784d 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -18707,6 +18707,78 @@ } } }, + "/api/v1/schemas/vocabulary": { + "get": { + "tags": [ + "schemas" + ], + "summary": "Get Schema Vocabulary", + "description": "The primitives, use cases, attributes and override types a column may use.", + "operationId": "schemas.vocabulary", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SchemaVocabulary" + } + } + } + }, + "401": { + "description": "No valid session: the credentials are missing, invalid or expired, or the account behind them is disabled or blocked", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + }, + "403": { + "description": "Refused: the session lacks the action this route checks, or its account must replace an issued password first. A SCIM route answers a refused action in the RFC 7644 error envelope", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + }, + "default": { + "description": "Problem details (RFC 9457)", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + } + }, + "security": [ + { + "BearerAuth": [] + } + ], + "x-scalo": { + "group": "schemas", + "resource": "schemas", + "hierarchy": [ + "schemas" + ], + "verb": "vocabulary", + "read_only": true, + "cli": { + "hidden": false, + "aliases": [], + "positional": [] + }, + "action": "schema:read" + } + } + }, "/api/v1/schemas/definitions/{schema_path}/versions/columns": { "get": { "tags": [ @@ -42132,6 +42204,11 @@ "deployment_default": { "type": "integer", "title": "Deployment Default" + }, + "maximum": { + "type": "integer", + "title": "Maximum", + "description": "The largest value a PATCH accepts, so a form can bound it." } }, "type": "object", @@ -42139,7 +42216,8 @@ "effective", "stored", "origin", - "deployment_default" + "deployment_default", + "maximum" ], "title": "IntDefault", "description": "One integer default: the value in force, and where it came from." @@ -50349,6 +50427,47 @@ "title": "SchemaVersionGet", "description": "The selected version's metadata, its columns paged in the answer's ``items``." }, + "SchemaVocabulary": { + "properties": { + "primitives": { + "items": { + "$ref": "#/components/schemas/VocabularyPrimitive" + }, + "type": "array", + "title": "Primitives" + }, + "use_cases": { + "items": { + "$ref": "#/components/schemas/VocabularyUseCase" + }, + "type": "array", + "title": "Use Cases" + }, + "attributes": { + "items": { + "$ref": "#/components/schemas/VocabularyAttribute" + }, + "type": "array", + "title": "Attributes" + }, + "ch_overrides": { + "items": { + "$ref": "#/components/schemas/VocabularyChOverride" + }, + "type": "array", + "title": "Ch Overrides" + } + }, + "type": "object", + "required": [ + "primitives", + "use_cases", + "attributes", + "ch_overrides" + ], + "title": "SchemaVocabulary", + "description": "What a schema form may offer, read from the dfe-schemas registry the engine runs." + }, "SelectionResponse": { "properties": { "rules": { @@ -55958,6 +56077,190 @@ "title": "ViewParameter", "description": "Parameter discovered from a ClickHouse parameterized view.\n\nExtracted from the view's CREATE SQL by parsing {param:Type} patterns.\nProvides type metadata across the full stack (ClickHouse -> Python -> TypeScript -> HTML)." }, + "VocabularyAttribute": { + "properties": { + "name": { + "type": "string", + "title": "Name" + }, + "label": { + "type": "string", + "title": "Label" + }, + "valid_primitives": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Valid Primitives" + } + }, + "type": "object", + "required": [ + "name", + "label", + "valid_primitives" + ], + "title": "VocabularyAttribute", + "description": "An attribute a column may carry." + }, + "VocabularyChOverride": { + "properties": { + "value": { + "type": "string", + "title": "Value" + }, + "label": { + "type": "string", + "title": "Label" + }, + "primitive": { + "type": "string", + "title": "Primitive" + }, + "parameters": { + "items": { + "$ref": "#/components/schemas/VocabularyOverrideParameter" + }, + "type": "array", + "title": "Parameters", + "description": "Inputs the type needs, in the order the type takes them." + }, + "template": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Template", + "description": "The override text with each parameter in braces; null when it takes none." + }, + "hint": { + "type": "string", + "title": "Hint", + "description": "A note to show beside the option, empty when there is none." + } + }, + "type": "object", + "required": [ + "value", + "label", + "primitive", + "parameters", + "template", + "hint" + ], + "title": "VocabularyChOverride", + "description": "A ClickHouse type offered as an override, and the primitive it refines." + }, + "VocabularyOverrideParameter": { + "properties": { + "name": { + "type": "string", + "title": "Name", + "description": "The name the template wraps in braces." + }, + "label": { + "type": "string", + "title": "Label" + }, + "kind": { + "type": "string", + "enum": [ + "integer", + "text" + ], + "title": "Kind" + } + }, + "type": "object", + "required": [ + "name", + "label", + "kind" + ], + "title": "VocabularyOverrideParameter", + "description": "An argument an override type takes, and the input a form shows for it." + }, + "VocabularyPrimitive": { + "properties": { + "name": { + "type": "string", + "title": "Name" + }, + "label": { + "type": "string", + "title": "Label" + }, + "ch_type": { + "type": "string", + "title": "Ch Type", + "description": "The ClickHouse type the primitive maps to." + } + }, + "type": "object", + "required": [ + "name", + "label", + "ch_type" + ], + "title": "VocabularyPrimitive", + "description": "A primitive a column may declare." + }, + "VocabularyUseCase": { + "properties": { + "name": { + "type": "string", + "title": "Name" + }, + "label": { + "type": "string", + "title": "Label" + }, + "description": { + "type": "string", + "title": "Description", + "description": "The question the use case answers, in plain words." + }, + "valid_primitives": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Valid Primitives" + }, + "parameters": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Parameters", + "description": "Arguments the use case takes in brackets, such as dims for similarity_search." + }, + "retired_names": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Retired Names", + "description": "Names this use case replaced. A stored column may still carry one." + } + }, + "type": "object", + "required": [ + "name", + "label", + "description", + "valid_primitives", + "parameters", + "retired_names" + ], + "title": "VocabularyUseCase", + "description": "An index use case: the question the user asks of a column." + }, "VolumeBand": { "type": "string", "enum": [ diff --git a/scripts/route-plan.toml b/scripts/route-plan.toml index 3f811c27c..c27aee0b0 100644 --- a/scripts/route-plan.toml +++ b/scripts/route-plan.toml @@ -222,6 +222,7 @@ routes = [ { old = "DELETE /api/v1/schemas/definitions/{schema_path}/versions/{version}", new = "DELETE /api/v1/schemas/{schema}/versions/{version}", status = "moved" }, { old = "POST /api/v1/schemas/elastic-converter", status = "unchanged" }, { old = "POST /api/v1/schemas/import", status = "unchanged" }, + { old = "GET /api/v1/schemas/vocabulary", status = "unchanged" }, { old = "POST /api/v1/schemas/{source_name}/build", status = "decide", note = "source-schema-overlap" }, { old = "GET /api/v1/schemas/{source_name}/columns", status = "decide", note = "source-schema-overlap" }, { old = "GET /api/v1/schemas/{source_name}/json-paths", new = "GET /api/v1/sources/{source}/json-paths", verb = "json-paths", status = "moved" }, diff --git a/src/dfe_engine/api/v1/schemas.py b/src/dfe_engine/api/v1/schemas.py index 3c5771351..081c0d91b 100644 --- a/src/dfe_engine/api/v1/schemas.py +++ b/src/dfe_engine/api/v1/schemas.py @@ -98,6 +98,7 @@ from dfe_engine.services.schema.json_promotion_service import PROMOTED_FIELD_TYPE from dfe_engine.source.models import Source, SourceHeader, SourceWriteRequest from dfe_engine.source.registry import SourceNotFoundError, SourceValidationError +from dfe_engine.source.type_registry import TypeRegistry router = APIRouter(prefix="/schemas", tags=["schemas"]) @@ -576,6 +577,93 @@ async def list_schemas( ) +class _FromRegistry(BaseModel): + """Built from the registry's frozen entries, which are not dicts.""" + + model_config = ConfigDict(from_attributes=True) + + +class VocabularyPrimitive(_FromRegistry): + """A primitive a column may declare.""" + + name: str + label: str + ch_type: str = Field(description="The ClickHouse type the primitive maps to.") + + +class VocabularyUseCase(_FromRegistry): + """An index use case: the question the user asks of a column.""" + + name: str + label: str + description: str = Field(description="The question the use case answers, in plain words.") + valid_primitives: list[str] + parameters: list[str] = Field( + description="Arguments the use case takes in brackets, such as dims for similarity_search." + ) + retired_names: list[str] = Field( + description="Names this use case replaced. A stored column may still carry one." + ) + + +class VocabularyAttribute(_FromRegistry): + """An attribute a column may carry.""" + + name: str + label: str + valid_primitives: list[str] + + +class VocabularyOverrideParameter(_FromRegistry): + """An argument an override type takes, and the input a form shows for it.""" + + name: str = Field(description="The name the template wraps in braces.") + label: str + kind: Literal["integer", "text"] + + +class VocabularyChOverride(_FromRegistry): + """A ClickHouse type offered as an override, and the primitive it refines.""" + + value: str + label: str + primitive: str + parameters: list[VocabularyOverrideParameter] = Field( + description="Inputs the type needs, in the order the type takes them." + ) + template: str | None = Field( + description="The override text with each parameter in braces; null when it takes none." + ) + hint: str = Field(description="A note to show beside the option, empty when there is none.") + + +class SchemaVocabulary(_FromRegistry): + """What a schema form may offer, read from the dfe-schemas registry the engine runs.""" + + primitives: list[VocabularyPrimitive] + use_cases: list[VocabularyUseCase] + attributes: list[VocabularyAttribute] + ch_overrides: list[VocabularyChOverride] + + +@router.get( + "/vocabulary", + response_model=SchemaVocabulary, + dependencies=[Depends(require_action(scopes_dict["schema_read"]))], +) +def get_schema_vocabulary(user: CurrentUser) -> SchemaVocabulary: + """The primitives, use cases, attributes and override types a column may use.""" + vocabulary = TypeRegistry.default().vocabulary() + return SchemaVocabulary.model_validate( + { + "primitives": vocabulary.primitives, + "use_cases": vocabulary.use_cases, + "attributes": vocabulary.attributes, + "ch_overrides": vocabulary.ch_override_options, + } + ) + + @router.get( "/definitions/{schema_path:path}/versions/columns", response_model=MetaSchemaGetResponse, diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py index dfa54b054..d4340bf0d 100644 --- a/src/dfe_engine/api/v1/sources.py +++ b/src/dfe_engine/api/v1/sources.py @@ -136,9 +136,9 @@ ttl_change_for_table, ) from dfe_engine.source.external import ( - HEADER_FIELDS, external_connection, external_group_lag, + header_fields, sample_record, ) from dfe_engine.source.flow import FlowError, SourceFlow, resolve_flow @@ -3790,7 +3790,7 @@ def _connection_model( allowed_sources=list(connection.allowed_sources), header_fields=[ HeaderFieldModel(field=f.field, when_missing=f.when_missing, when_set=f.when_set) - for f in HEADER_FIELDS + for f in header_fields() ], sample_record=sample_record(source), ) diff --git a/src/dfe_engine/api/v1/system.py b/src/dfe_engine/api/v1/system.py index 2d8ddd343..fe46c4b3a 100644 --- a/src/dfe_engine/api/v1/system.py +++ b/src/dfe_engine/api/v1/system.py @@ -674,6 +674,7 @@ class IntDefault(BaseModel): stored: int | None = Field(description="The admin's override; null when none is stored.") origin: Literal["override", "deployment"] deployment_default: int + maximum: int = Field(description="The largest value a PATCH accepts, so a form can bound it.") class StrDefault(BaseModel): @@ -751,6 +752,7 @@ def _system_defaults(request: Request, settings: Any) -> SystemDefaults: stored=ttl.stored, origin=ttl.origin, deployment_default=ttl.deployment_default, + maximum=MAX_DEFAULT_TTL_DAYS, ), common_header_type=StrDefault( effective=table.header_type, diff --git a/src/dfe_engine/source/external.py b/src/dfe_engine/source/external.py index 84e697c52..2b72775a7 100644 --- a/src/dfe_engine/source/external.py +++ b/src/dfe_engine/source/external.py @@ -15,7 +15,7 @@ keeps only the reference the operator gives it. The loader fills in a missing common header field as it does for any producer, -and :data:`HEADER_FIELDS` says what each one costs when it is missing. It holds +and :func:`header_fields` says what each one costs when it is missing. It holds ``_org_id`` and ``_source`` on the pipeline's topic to the source's own org and name plus the source's allow-lists, so a pipeline cannot write into another tenant's data or another source's table. @@ -23,11 +23,20 @@ See docs/data-plane/external-pipelines.md. """ +import functools +import re from dataclasses import dataclass from typing import TYPE_CHECKING, Literal from dfe_engine.kafka.topics import TopicAdmin, consumer_group_backlog -from dfe_engine.source.models import Source, external_consumer_group, transformed_topic +from dfe_engine.schema.schema_loader import SchemaLoader +from dfe_engine.source.models import ( + DEFAULT_HEADER_TYPE, + DEFAULT_HEADER_VERSION, + Source, + external_consumer_group, + transformed_topic, +) if TYPE_CHECKING: from dfe_engine.settings import DFESettings @@ -44,43 +53,94 @@ class HeaderField: when_set: str -HEADER_FIELDS: tuple[HeaderField, ...] = ( - HeaderField( - "_timestamp", - "the loader reads timestamp, then @timestamp, and takes the load time when " - "neither is there", - "the event's own time, so time-range queries and retention follow when it happened", - ), - HeaderField( - "_org_id", - "the record lands in the source's org", +# What a value the pipeline sets buys. The header declares where a field is read +# from and not why that is worth doing, so this is the one piece of prose kept here. +_WHEN_SET: dict[str, str] = { + "_timestamp": "the event's own time, so time-range queries and retention follow when it happened", + "_org_id": ( "the source's org, or one allowed_org_ids lists, is kept, which is how a provider " "lands each customer's records in that customer's org. Any other value is " - "replaced with the source's org and counted", + "replaced with the source's org and counted" ), - HeaderField( - "_source", - "the loader takes it from the topic name, which is this source", + "_source": ( "this source, or one allowed_sources lists, is kept and picks the table. Any other " - "value is replaced with this source and counted", - ), - HeaderField( - "_tags", - "the loader reads tags, then meta, then metadata.tags, and leaves it empty " - "when none is there", - "tags a query or an alert can filter on without parsing the payload", + "value is replaced with this source and counted" ), - HeaderField( - "_timestamp_received", - "the loader takes the load time", - "when the pipeline first saw the event, so the delay before DFE can be measured", + "_tags": "tags a query or an alert can filter on without parsing the payload", + "_timestamp_received": ( + "when the pipeline first saw the event, so the delay before DFE can be measured" ), +} + +# The fallbacks the header's expressions name, in the words a customer reads. +_FALLBACKS: dict[str, str] = { + "now()": "it takes the load time", + "topic_name": "it takes the topic name, which is this source", +} + +# What the loader does when none of a field's candidates is there and the +# expression names no fallback of its own. The loader owns these, not the header. +_WHEN_NONE: dict[str, str] = { + "_org_id": "the record lands in the source's org", + "_tags": "it stays empty", + "_timestamp_received": "it takes the load time", +} + +_SOURCE_EXPR = re.compile( + r"^@source:\s*(?:first\((?P[^)]+)\)|(?P[^|\s][^|]*?))\s*(?:\|\s*(?P.+?))?\s*$" ) -"""The fields the default header profile reads from a record, in dfe-schemas column order. -``_timestamp_load``, ``_uuid``, ``_json`` and ``_raw`` are left out: the loader -and ClickHouse always write those, and a value the pipeline sets is not read. -""" + +def reads_from_expr(field: str, expr: str) -> str | None: + """What the loader does with a header column's ``@source`` expression, in words. + + ``@source: timestamp | now()`` reads "the loader reads timestamp, and when it is + not there it takes the load time". An expression that is not an ``@source`` + returns None, because the loader writes that column itself. + """ + match = _SOURCE_EXPR.match(expr.strip()) + if match is None: + return None + first = match.group("first") + candidates = ( + [c.strip() for c in first.split("/") if c.strip()] + if first + else [match.group("field").strip()] + ) + text = "the loader reads " + ", then ".join(candidates) + fallback = match.group("fallback") + outcome = _FALLBACKS.get(fallback.strip(), f"it uses {fallback.strip()}") if fallback else None + outcome = outcome or _WHEN_NONE.get(field) + if outcome: + absent = "it is not there" if len(candidates) == 1 else "none is there" + text += f", and when {absent} {outcome}" + return text + + +@functools.cache +def header_fields( + header_type: str = DEFAULT_HEADER_TYPE, header_version: str = DEFAULT_HEADER_VERSION +) -> tuple[HeaderField, ...]: + """The fields the header profile reads from a record, in its column order. + + Read from the pinned dfe-schemas profile, so the read order named here is the + one the loader applies. ``_timestamp_load``, ``_uuid``, ``_json`` and ``_raw`` + carry no ``@source`` expression: the loader and ClickHouse always write those, + and a value the pipeline sets is not read. + """ + fields = [] + for column in SchemaLoader.load_profile(header_type, profile_version=header_version): + reads = reads_from_expr(column.name, column.expr or "") + if reads is None: + continue + fields.append( + HeaderField( + field=column.name, + when_missing=reads, + when_set=_WHEN_SET.get(column.name, "the pipeline's own value"), + ) + ) + return tuple(fields) @dataclass(frozen=True, slots=True) @@ -143,7 +203,7 @@ def external_connection( def sample_record(source: Source) -> dict[str, object]: - """A record carrying every field in :data:`HEADER_FIELDS`, for the pipeline to copy.""" + """A record carrying every field in :func:`header_fields`, for the pipeline to copy.""" return { "_timestamp": "2026-11-03T04:05:06.789Z", "_org_id": "", diff --git a/src/dfe_engine/source/type_registry.py b/src/dfe_engine/source/type_registry.py index c66956814..060471399 100644 --- a/src/dfe_engine/source/type_registry.py +++ b/src/dfe_engine/source/type_registry.py @@ -126,6 +126,82 @@ def current_use_case(declared: Any) -> Any: return RETIRED_USE_CASES.get(declared.strip(), declared) +@dataclass(frozen=True) +class PrimitiveEntry: + """One primitive a column may declare, as a form offers it.""" + + name: str + label: str + ch_type: str + + +@dataclass(frozen=True) +class UseCaseEntry: + """One index use case: the question the user asks of a column.""" + + name: str + label: str + description: str + valid_primitives: tuple[str, ...] + parameters: tuple[str, ...] + retired_names: tuple[str, ...] + + +@dataclass(frozen=True) +class AttributeEntry: + """One attribute a column may carry, with the primitives it applies to.""" + + name: str + label: str + valid_primitives: tuple[str, ...] + + +@dataclass(frozen=True) +class OverrideParameter: + """One argument an override type takes, and the input a form shows for it.""" + + name: str + label: str + kind: Literal["integer", "text"] + + +@dataclass(frozen=True) +class ChOverrideOption: + """One ClickHouse type a form offers as an override, and the primitive it refines. + + ``template`` is the override text with each parameter name in braces + (``Decimal({precision},{scale})``). It is None for a type that takes none. + """ + + value: str + label: str + primitive: str + parameters: tuple[OverrideParameter, ...] + template: str | None + hint: str + + +@dataclass(frozen=True) +class Vocabulary: + """What a schema form may offer, read from the registry.""" + + primitives: tuple[PrimitiveEntry, ...] + use_cases: tuple[UseCaseEntry, ...] + attributes: tuple[AttributeEntry, ...] + ch_override_options: tuple[ChOverrideOption, ...] + + +def default_label(name: str) -> str: + """A readable label from a registry name, for an entry that declares none.""" + spaced = name.replace("_", " ").strip() + return spaced[:1].upper() + spaced[1:] + + +def _text(entry: dict[str, Any], key: str, default: str = "") -> str: + value = entry.get(key) + return value.strip() if isinstance(value, str) and value.strip() else default + + class TypeRegistry: """Canonical registry mapping primitives to ClickHouse types. @@ -395,3 +471,102 @@ def primitive_defaults(self, primitive: str) -> dict[str, Any]: if primitive not in self._primitives: raise UnknownPrimitiveError(f"Unknown primitive '{primitive}'") return dict(self._primitives[primitive]) + + # ----------------------------------------------------------------- + # Vocabulary served to the schema forms + # ----------------------------------------------------------------- + + def vocabulary(self) -> Vocabulary: + """Everything a schema form may offer, from this registry. + + A label the registry does not declare reads off the entry's name. The + retired ``lowcardinality`` attribute is left out, because a column now + declares ``cardinality``. Each use case carries the names it replaced, so + a stored column that still holds one resolves to it. + + Raises: + TypeRegistryError: An override option names no value, or a primitive + the registry does not have. + """ + names = tuple(self._primitives) + retired: dict[str, list[str]] = {} + for old, current in RETIRED_USE_CASES.items(): + retired.setdefault(current, []).append(old) + return Vocabulary( + primitives=tuple( + PrimitiveEntry( + name=name, + label=_text(entry, "label", default_label(name)), + ch_type=_text(entry, "ch_type"), + ) + for name, entry in self._primitives.items() + ), + use_cases=tuple( + UseCaseEntry( + name=name, + label=_text(entry, "label", default_label(name)), + description=_text(entry, "description"), + valid_primitives=tuple(entry.get("valid_primitives") or ()), + parameters=tuple(entry.get("parameters") or ()), + retired_names=tuple(retired.get(name, ())), + ) + for name, entry in self._use_cases.items() + ), + attributes=tuple( + AttributeEntry( + name=name, + label=_text(entry, "label", default_label(name)), + valid_primitives=names + if entry.get("valid_primitives") == "all" + else tuple(entry.get("valid_primitives") or ()), + ) + for name, entry in self._attributes.items() + if name != LOWCARDINALITY_ATTRIBUTE + ), + ch_override_options=tuple( + self._override_option(raw) for raw in self._ch_overrides.get("options") or () + ), + ) + + def _override_option(self, raw: Any) -> ChOverrideOption: + if not isinstance(raw, dict) or not _text(raw, "value"): + raise TypeRegistryError("a ch_overrides option must be a mapping with a 'value'") + value = _text(raw, "value") + primitive = _text(raw, "primitive") + if primitive not in self._primitives: + raise TypeRegistryError( + f"ch_overrides option '{value}' names primitive '{primitive}', " + f"which the registry does not have" + ) + template = _text(raw, "template") or None + parameters = tuple(self._override_parameter(value, p) for p in raw.get("parameters") or ()) + if bool(parameters) != (template is not None): + raise TypeRegistryError( + f"ch_overrides option '{value}' must give both 'parameters' and 'template', or neither" + ) + for parameter in parameters: + if "{" + parameter.name + "}" not in (template or ""): + raise TypeRegistryError( + f"ch_overrides option '{value}' template does not use '{{{parameter.name}}}'" + ) + return ChOverrideOption( + value=value, + label=_text(raw, "label", value), + primitive=primitive, + parameters=parameters, + template=template, + hint=_text(raw, "hint"), + ) + + @staticmethod + def _override_parameter(value: str, raw: Any) -> OverrideParameter: + name = _text(raw, "name") if isinstance(raw, dict) else "" + kind = _text(raw, "kind") if isinstance(raw, dict) else "" + if not name or kind not in ("integer", "text"): + raise TypeRegistryError( + f"ch_overrides option '{value}' has a parameter without a name " + f"and a kind of integer or text" + ) + return OverrideParameter( + name=name, label=_text(raw, "label", default_label(name)), kind=kind + ) diff --git a/tests/unit/test_api/test_schemas_router.py b/tests/unit/test_api/test_schemas_router.py index eab9946b7..285a8798b 100644 --- a/tests/unit/test_api/test_schemas_router.py +++ b/tests/unit/test_api/test_schemas_router.py @@ -15,6 +15,7 @@ from types import SimpleNamespace +from dfe_engine.source.type_registry import TypeRegistry from tests.support.accounts import OnboardedClient, session_token from tests.support.present import present @@ -23,6 +24,44 @@ ADMIN_PASSWORD = "test-admin-pw" +class TestSchemaVocabulary: + """GET /api/v1/schemas/vocabulary serves what the schema forms offer.""" + + URL = "/api/v1/schemas/vocabulary" + + def test_serves_every_registry_entry_with_a_label(self, client, admin_headers): + resp = client.get(self.URL, headers=admin_headers) + + assert resp.status_code == 200, resp.text + body = resp.json() + registry = TypeRegistry.default() + assert {p["name"] for p in body["primitives"]} == set(registry.primitives) + assert {u["name"] for u in body["use_cases"]} == set(registry.use_cases) + for entry in body["primitives"] + body["use_cases"] + body["attributes"]: + assert entry["label"], entry + + def test_a_use_case_carries_its_arguments_and_the_names_it_replaced( + self, client, admin_headers + ): + body = client.get(self.URL, headers=admin_headers).json() + use_cases = {u["name"]: u for u in body["use_cases"]} + + assert use_cases["similarity_search"]["parameters"] == ["dims"] + assert use_cases["similarity_search"]["valid_primitives"] == ["vector"] + assert use_cases["word_search"]["retired_names"] == ["fulltext"] + assert use_cases["exact_match"]["retired_names"] == ["bloom"] + assert not {"fulltext", "bloom"} & set(use_cases) + + def test_the_retired_lowcardinality_attribute_is_not_offered(self, client, admin_headers): + body = client.get(self.URL, headers=admin_headers).json() + + assert "lowcardinality" not in {a["name"] for a in body["attributes"]} + assert {a["name"] for a in body["attributes"]} >= {"nullable", "not_null"} + + def test_requires_authentication(self, client): + assert client.get(self.URL).status_code == 401 + + class TestSchemasRouter: """GET/POST /api/v1/sources/{name}/columns and /build (formerly under /schemas).""" diff --git a/tests/unit/test_api/test_system_defaults.py b/tests/unit/test_api/test_system_defaults.py index 7274b91b4..59b87adfd 100644 --- a/tests/unit/test_api/test_system_defaults.py +++ b/tests/unit/test_api/test_system_defaults.py @@ -21,6 +21,7 @@ from dfe_engine.api.deps import _registries from dfe_engine.clickhouse.clickhouse_manager import ClickHouseManager from dfe_engine.gitcrud import GitCrud, default_registry +from dfe_engine.gitcrud.retention import MAX_DEFAULT_TTL_DAYS from dfe_engine.gitops.repo import GitopsRepo from dfe_engine.settings import ClickHouseResilienceSettings, ClickHouseSettings @@ -40,6 +41,7 @@ def _deployment_body(*, editable: bool) -> dict: "stored": None, "origin": "deployment", "deployment_default": 90, + "maximum": MAX_DEFAULT_TTL_DAYS, }, "common_header_type": { "effective": "timeseries", diff --git a/tests/unit/test_source/test_external_pipeline.py b/tests/unit/test_source/test_external_pipeline.py index 1346ff4c4..7da29ae6b 100644 --- a/tests/unit/test_source/test_external_pipeline.py +++ b/tests/unit/test_source/test_external_pipeline.py @@ -24,6 +24,7 @@ from dfe_engine.orgs.tenancy import Tenancy from dfe_engine.services.source_routing import compile_loader_routing, version_rules from dfe_engine.settings import DFESettings +from dfe_engine.source.external import header_fields, reads_from_expr from dfe_engine.source.flow import FlowError, resolve_flow from dfe_engine.source.models import Source, SourceTransform, SourceWriteRequest from tests.unit.test_services.test_source_routing import FakeSourceRegistry @@ -279,7 +280,13 @@ def test_a_transform_is_told_what_to_read_write_and_join(self): assert block.bootstrap == "kafka.example:9094" assert block.credential_ref == "vault:kv/acme/kafka" assert block.allowed_sources == ["acme-audit"] - assert [field.field for field in block.header_fields][:2] == ["_timestamp", "_org_id"] + assert [field.field for field in block.header_fields] == [ + "_timestamp", + "_timestamp_received", + "_org_id", + "_source", + "_tags", + ] assert block.sample_record["_source"] == "acme-logs" def test_an_origin_writes_the_landing_topic_and_joins_no_group(self): @@ -307,3 +314,48 @@ def test_a_source_with_no_external_pipeline_has_no_block(self): receiver = Source.model_validate({"source": "auth", "match": {"field": "a", "value": "b"}}) assert _connection_model(receiver, _settings()) is None + + +@pytest.mark.parametrize( + ("field", "expr", "text"), + [ + ( + "_timestamp", + "@source: timestamp | now()", + "the loader reads timestamp, and when it is not there it takes the load time", + ), + ( + "_tags", + "@source: first(_tags/tags/meta/metadata.tags)", + "the loader reads _tags, then tags, then meta, then metadata.tags, " + "and when none is there it stays empty", + ), + ( + "_source", + "@source: first(_source) | topic_name", + "the loader reads _source, and when it is not there " + "it takes the topic name, which is this source", + ), + ( + "other", + "@source: a | lower(a)", + "the loader reads a, and when it is not there it uses lower(a)", + ), + ("other", "@source: first(a/b)", "the loader reads a, then b"), + ("_uuid", "@generated: generateUUIDv7()", None), + ("_raw", "", None), + ], +) +def test_what_the_loader_reads_comes_from_the_header_expression(field, expr, text): + assert reads_from_expr(field, expr) == text + + +def test_the_fields_a_pipeline_can_set_are_the_default_headers_source_columns(): + fields = {f.field: f.when_missing for f in header_fields()} + + assert list(fields) == ["_timestamp", "_timestamp_received", "_org_id", "_source", "_tags"] + assert fields["_timestamp"] == ( + "the loader reads timestamp, and when it is not there it takes the load time" + ) + assert "@timestamp" not in fields["_timestamp"] + assert fields["_tags"].startswith("the loader reads _tags, then tags") diff --git a/tests/unit/test_source/test_type_registry.py b/tests/unit/test_source/test_type_registry.py index f90996f71..e5c43c275 100644 --- a/tests/unit/test_source/test_type_registry.py +++ b/tests/unit/test_source/test_type_registry.py @@ -397,3 +397,137 @@ def test_primitive_defaults(self, registry: TypeRegistry): def test_primitive_defaults_unknown(self, registry: TypeRegistry): with pytest.raises(UnknownPrimitiveError): registry.primitive_defaults("nonexistent") + + +_DECLARED = { + "primitives": { + "string": {"ch_type": "String", "label": "Text line"}, + "decimalish": {"ch_type": "Float64"}, + "map": {"ch_type": "Map(String,String)"}, + }, + "use_cases": { + "word_search": { + "valid_primitives": ["string"], + "label": "Word search", + "description": "I search for whole words", + }, + "similarity_search": {"valid_primitives": ["string"], "parameters": ["dims"]}, + "bloom": {"valid_primitives": ["string"]}, + }, + "attributes": { + "lowcardinality": {"valid_primitives": ["string"]}, + "not_null": {"valid_primitives": "all", "label": "Not null"}, + }, + "ch_overrides": { + "exact": ["Map(String,String)"], + "options": [ + {"value": "Int8", "label": "Int8", "primitive": "string"}, + { + "value": "Decimal", + "label": "Decimal (fixed)", + "primitive": "decimalish", + "parameters": [ + {"name": "precision", "kind": "integer"}, + {"name": "scale", "label": "Scale places", "kind": "integer"}, + ], + "template": "Decimal({precision},{scale})", + "hint": "Fixed point.", + }, + ], + }, +} + + +class TestVocabulary: + def test_a_declared_label_wins_and_a_missing_one_reads_off_the_name(self): + vocabulary = TypeRegistry(_DECLARED).vocabulary() + + assert {p.name: p.label for p in vocabulary.primitives} == { + "string": "Text line", + "decimalish": "Decimalish", + "map": "Map", + } + use_cases = {u.name: u for u in vocabulary.use_cases} + assert use_cases["word_search"].description == "I search for whole words" + assert use_cases["similarity_search"].label == "Similarity search" + assert use_cases["similarity_search"].parameters == ("dims",) + + def test_a_use_case_lists_the_names_it_replaced(self): + use_cases = {u.name: u for u in TypeRegistry(_DECLARED).vocabulary().use_cases} + + assert use_cases["word_search"].retired_names == ("fulltext",) + + def test_the_retired_attribute_is_left_out_and_all_means_every_primitive(self): + attributes = {a.name: a for a in TypeRegistry(_DECLARED).vocabulary().attributes} + + assert set(attributes) == {"not_null"} + assert attributes["not_null"].valid_primitives == ("string", "decimalish", "map") + + def test_override_options_keep_their_template_and_hint(self): + options = {o.value: o for o in TypeRegistry(_DECLARED).vocabulary().ch_override_options} + + assert options["Int8"].template is None + assert options["Int8"].parameters == () + assert options["Decimal"].template == "Decimal({precision},{scale})" + assert [(p.name, p.label, p.kind) for p in options["Decimal"].parameters] == [ + ("precision", "Precision", "integer"), + ("scale", "Scale places", "integer"), + ] + assert options["Decimal"].hint == "Fixed point." + + @pytest.mark.parametrize( + ("option", "message"), + [ + ({"label": "No value", "primitive": "string"}, "with a 'value'"), + ({"value": "Int8", "primitive": "nope"}, "primitive 'nope'"), + ({"value": "Int8", "primitive": "string", "template": "Int8({n})"}, "or neither"), + ( + { + "value": "Int8", + "primitive": "string", + "parameters": [{"name": "n", "kind": "integer"}], + }, + "or neither", + ), + ( + { + "value": "Int8", + "primitive": "string", + "template": "Int8({n})", + "parameters": [{"name": "n", "kind": "float"}], + }, + "kind of integer or text", + ), + ( + { + "value": "Int8", + "primitive": "string", + "template": "Int8()", + "parameters": [{"name": "n", "kind": "integer"}], + }, + "does not use", + ), + ], + ) + def test_a_malformed_override_option_is_refused(self, option, message): + data = {**_DECLARED, "ch_overrides": {"options": [option]}} + + with pytest.raises(TypeRegistryError, match=message): + TypeRegistry(data).vocabulary() + + def test_the_shipped_registry_offers_only_overrides_it_accepts(self, registry: TypeRegistry): + for option in registry.vocabulary().ch_override_options: + registry.validate_ch_override( + (option.template or option.value).format( + **{parameter.name: 8 for parameter in option.parameters} + ) + ) + + def test_the_shipped_registry_labels_every_entry(self, registry: TypeRegistry): + vocabulary = registry.vocabulary() + + assert len(vocabulary.primitives) == len(registry.primitives) + assert all( + entry.label + for entry in (*vocabulary.primitives, *vocabulary.use_cases, *vocabulary.attributes) + ) From 4e88fa1d419504de2d1268a970ff2683961003ba Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 18:13:00 +1100 Subject: [PATCH 2/7] fix(schemas): serve the cardinality vocabulary The form needs to offer `cardinality: low` before a column can use it. The vocabulary dropped the retired `lowcardinality` attribute without its replacement. The vocabulary has `cardinality` values, default and labels. It also lists what each retired attribute reads as (`lowcardinality` is `low`). The vocabulary reads from the `cardinality` block of dfe-schemas types.yaml. Labels come from its `labels` map when present. --- docs/data-plane/schema.md | 2 +- openapi/openapi.json | 45 ++++++++++++++++++++ src/dfe_engine/api/v1/schemas.py | 13 ++++++ src/dfe_engine/source/type_registry.py | 19 +++++++++ tests/unit/test_api/test_schemas_router.py | 10 +++++ tests/unit/test_source/test_type_registry.py | 28 +++++++++++- 6 files changed, 115 insertions(+), 2 deletions(-) diff --git a/docs/data-plane/schema.md b/docs/data-plane/schema.md index 2253cf873..1d7598b3b 100644 --- a/docs/data-plane/schema.md +++ b/docs/data-plane/schema.md @@ -373,7 +373,7 @@ catalogue: | **Enum** | `Enum8(...)`, `Enum16(...)` | | **Special** | `Dynamic`, `Variant(...)`, `AggregateFunction(...)`, `SimpleAggregateFunction(...)` | -`GET /api/v1/schemas/vocabulary` serves what a schema form offers, read from this registry: the primitives, the index use cases with their labels and the names they replaced, the attributes, and the override types with the primitive each refines. A type that takes parameters carries a `template` such as `Decimal({precision},{scale})`, so a form builds the override text without knowing the type. A label the registry does not declare reads off the entry's name. The retired `lowcardinality` attribute is not offered: a column declares `cardinality`. +`GET /api/v1/schemas/vocabulary` serves what a schema form offers, read from this registry: the primitives, the index use cases with their labels and the names they replaced, the attributes, the cardinalities with their default, and the override types with the primitive each refines. A type that takes parameters carries a `template` such as `Decimal({precision},{scale})`, so a form builds the override text without knowing the type. A label the registry does not declare reads off the entry's name. The retired `lowcardinality` attribute is not offered: a column declares `cardinality`, and the vocabulary says a stored `lowcardinality` reads as `low`. --- diff --git a/openapi/openapi.json b/openapi/openapi.json index c821d784d..b597cb225 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -50429,6 +50429,9 @@ }, "SchemaVocabulary": { "properties": { + "cardinality": { + "$ref": "#/components/schemas/VocabularyCardinality" + }, "primitives": { "items": { "$ref": "#/components/schemas/VocabularyPrimitive" @@ -50460,6 +50463,7 @@ }, "type": "object", "required": [ + "cardinality", "primitives", "use_cases", "attributes", @@ -56104,6 +56108,47 @@ "title": "VocabularyAttribute", "description": "An attribute a column may carry." }, + "VocabularyCardinality": { + "properties": { + "values": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Values" + }, + "default": { + "type": "string", + "title": "Default", + "description": "What a column that declares none is read as." + }, + "labels": { + "additionalProperties": { + "type": "string" + }, + "type": "object", + "title": "Labels", + "description": "The label to show for each value." + }, + "retired_attributes": { + "additionalProperties": { + "type": "string" + }, + "type": "object", + "title": "Retired Attributes", + "description": "A retired attribute and the cardinality a column holding it reads as." + } + }, + "type": "object", + "required": [ + "values", + "default", + "labels", + "retired_attributes" + ], + "title": "VocabularyCardinality", + "description": "The cardinalities a column may declare." + }, "VocabularyChOverride": { "properties": { "value": { diff --git a/src/dfe_engine/api/v1/schemas.py b/src/dfe_engine/api/v1/schemas.py index 081c0d91b..969e1337c 100644 --- a/src/dfe_engine/api/v1/schemas.py +++ b/src/dfe_engine/api/v1/schemas.py @@ -637,9 +637,21 @@ class VocabularyChOverride(_FromRegistry): hint: str = Field(description="A note to show beside the option, empty when there is none.") +class VocabularyCardinality(_FromRegistry): + """The cardinalities a column may declare.""" + + values: list[str] + default: str = Field(description="What a column that declares none is read as.") + labels: dict[str, str] = Field(description="The label to show for each value.") + retired_attributes: dict[str, str] = Field( + description="A retired attribute and the cardinality a column holding it reads as." + ) + + class SchemaVocabulary(_FromRegistry): """What a schema form may offer, read from the dfe-schemas registry the engine runs.""" + cardinality: VocabularyCardinality primitives: list[VocabularyPrimitive] use_cases: list[VocabularyUseCase] attributes: list[VocabularyAttribute] @@ -656,6 +668,7 @@ def get_schema_vocabulary(user: CurrentUser) -> SchemaVocabulary: vocabulary = TypeRegistry.default().vocabulary() return SchemaVocabulary.model_validate( { + "cardinality": vocabulary.cardinality, "primitives": vocabulary.primitives, "use_cases": vocabulary.use_cases, "attributes": vocabulary.attributes, diff --git a/src/dfe_engine/source/type_registry.py b/src/dfe_engine/source/type_registry.py index 060471399..c0275c1d3 100644 --- a/src/dfe_engine/source/type_registry.py +++ b/src/dfe_engine/source/type_registry.py @@ -181,10 +181,21 @@ class ChOverrideOption: hint: str +@dataclass(frozen=True) +class CardinalityVocabulary: + """The cardinalities a column may declare, and how the retired spelling maps onto them.""" + + values: tuple[str, ...] + default: str + labels: dict[str, str] + retired_attributes: dict[str, str] + + @dataclass(frozen=True) class Vocabulary: """What a schema form may offer, read from the registry.""" + cardinality: CardinalityVocabulary primitives: tuple[PrimitiveEntry, ...] use_cases: tuple[UseCaseEntry, ...] attributes: tuple[AttributeEntry, ...] @@ -492,7 +503,15 @@ def vocabulary(self) -> Vocabulary: retired: dict[str, list[str]] = {} for old, current in RETIRED_USE_CASES.items(): retired.setdefault(current, []).append(old) + values = tuple(self.cardinalities) + labels = self._cardinality.get("labels") or {} return Vocabulary( + cardinality=CardinalityVocabulary( + values=values, + default=str(self._cardinality.get("default") or DEFAULT_CARDINALITY), + labels={name: _text(labels, name, default_label(name)) for name in values}, + retired_attributes={LOWCARDINALITY_ATTRIBUTE: LOW_CARDINALITY}, + ), primitives=tuple( PrimitiveEntry( name=name, diff --git a/tests/unit/test_api/test_schemas_router.py b/tests/unit/test_api/test_schemas_router.py index 285a8798b..272c4a0f9 100644 --- a/tests/unit/test_api/test_schemas_router.py +++ b/tests/unit/test_api/test_schemas_router.py @@ -52,6 +52,16 @@ def test_a_use_case_carries_its_arguments_and_the_names_it_replaced( assert use_cases["exact_match"]["retired_names"] == ["bloom"] assert not {"fulltext", "bloom"} & set(use_cases) + def test_cardinality_is_offered_with_its_default_and_the_retired_spelling( + self, client, admin_headers + ): + body = client.get(self.URL, headers=admin_headers).json() + + assert body["cardinality"]["values"] == ["low", "high", "unknown"] + assert body["cardinality"]["default"] == "unknown" + assert set(body["cardinality"]["labels"]) == {"low", "high", "unknown"} + assert body["cardinality"]["retired_attributes"] == {"lowcardinality": "low"} + def test_the_retired_lowcardinality_attribute_is_not_offered(self, client, admin_headers): body = client.get(self.URL, headers=admin_headers).json() diff --git a/tests/unit/test_source/test_type_registry.py b/tests/unit/test_source/test_type_registry.py index e5c43c275..761ff61f7 100644 --- a/tests/unit/test_source/test_type_registry.py +++ b/tests/unit/test_source/test_type_registry.py @@ -414,6 +414,11 @@ def test_primitive_defaults_unknown(self, registry: TypeRegistry): "similarity_search": {"valid_primitives": ["string"], "parameters": ["dims"]}, "bloom": {"valid_primitives": ["string"]}, }, + "cardinality": { + "values": ["low", "high", "unknown"], + "default": "unknown", + "labels": {"low": "Few values"}, + }, "attributes": { "lowcardinality": {"valid_primitives": ["string"]}, "not_null": {"valid_primitives": "all", "label": "Not null"}, @@ -452,6 +457,21 @@ def test_a_declared_label_wins_and_a_missing_one_reads_off_the_name(self): assert use_cases["similarity_search"].label == "Similarity search" assert use_cases["similarity_search"].parameters == ("dims",) + def test_cardinality_carries_its_labels_default_and_the_retired_spelling(self): + cardinality = TypeRegistry(_DECLARED).vocabulary().cardinality + + assert cardinality.values == ("low", "high", "unknown") + assert cardinality.default == "unknown" + assert cardinality.labels == {"low": "Few values", "high": "High", "unknown": "Unknown"} + assert cardinality.retired_attributes == {"lowcardinality": "low"} + + def test_a_registry_that_declares_no_cardinality_gets_the_package_default(self): + data = {k: v for k, v in _DECLARED.items() if k != "cardinality"} + + cardinality = TypeRegistry(data).vocabulary().cardinality + + assert (cardinality.values, cardinality.default) == (("low", "high", "unknown"), "unknown") + def test_a_use_case_lists_the_names_it_replaced(self): use_cases = {u.name: u for u in TypeRegistry(_DECLARED).vocabulary().use_cases} @@ -516,13 +536,19 @@ def test_a_malformed_override_option_is_refused(self, option, message): TypeRegistry(data).vocabulary() def test_the_shipped_registry_offers_only_overrides_it_accepts(self, registry: TypeRegistry): - for option in registry.vocabulary().ch_override_options: + options = registry.vocabulary().ch_override_options + + assert options + for option in options: registry.validate_ch_override( (option.template or option.value).format( **{parameter.name: 8 for parameter in option.parameters} ) ) + def test_the_shipped_registry_describes_every_use_case(self, registry: TypeRegistry): + assert all(entry.description for entry in registry.vocabulary().use_cases) + def test_the_shipped_registry_labels_every_entry(self, registry: TypeRegistry): vocabulary = registry.vocabulary() From 8e88346b37b5a2612ffcea3291433054c1393ac7 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 18:17:29 +1100 Subject: [PATCH 3/7] fix(schemas): whitelist the vocabulary fields vulture misses The vocabulary route reads these registry fields through pydantic from_attributes, which leaves no attribute access in the source for vulture to see. --- src/vulture_whitelist.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/vulture_whitelist.py b/src/vulture_whitelist.py index 8dbf8ec0f..9ffbe2154 100644 --- a/src/vulture_whitelist.py +++ b/src/vulture_whitelist.py @@ -118,3 +118,6 @@ def __getattr__(self, name: str) -> None: ... _.get_draft, _.get_published, _.get_version, _.list_adapters, _.load_clickhouse_settings _.load_plan, _.protocol_paths, _.query_names, _.read_payload, _.save_draft, _.save_plan _.set_deployed + +# Registry entries the schema vocabulary route reads through pydantic from_attributes. +_.retired_names, _.retired_attributes, _.ch_overrides From 93be6584292714f225ce5003ad7d2d697aa045f8 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 18:22:13 +1100 Subject: [PATCH 4/7] fix(schemas): move to dfe-schemas 0.2.29 0.2.29 has the type vocabulary labels and override options. The vocabulary route serves labels and 29 override types instead of names alone. --- pyproject.toml | 2 +- uv.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 84070d6ed..32ed07970 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,7 +82,7 @@ dependencies = [ # wheel. Floor is the first release that ships the source catalogue under # catalogue/. uv.lock pins the hash. # Upper bound: <1 is the API-stability boundary while dfe-schemas is 0.x. - "dfe-schemas>=0.2.28,<1", + "dfe-schemas>=0.2.29,<1", # Data processing "pandas>=3.0.3", "numpy>=2.5.4", diff --git a/uv.lock b/uv.lock index 147f13be0..50dbf37c5 100644 --- a/uv.lock +++ b/uv.lock @@ -714,7 +714,7 @@ requires-dist = [ { name = "confluent-kafka", specifier = ">=2.16.0" }, { name = "croniter", specifier = ">=6.2.2" }, { name = "cryptography", specifier = ">=50.0.0" }, - { name = "dfe-schemas", specifier = ">=0.2.28,<1" }, + { name = "dfe-schemas", specifier = ">=0.2.29,<1" }, { name = "dulwich", specifier = ">=1.2.7" }, { name = "faker", specifier = ">=40.43.0" }, { name = "fastapi", specifier = ">=0.143.0" }, @@ -769,14 +769,14 @@ dev = [ [[package]] name = "dfe-schemas" -version = "0.2.28" +version = "0.2.29" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "ruamel-yaml" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/88/86/6008855128ceebd39f37c00050b72254b7069839eec952fed4e3a1836650/dfe_schemas-0.2.28.tar.gz", hash = "sha256:1ea9691a306ee39b55d33f7d7d3d018f13e5b335718870e6dd20ef2de51d6964", size = 1293635, upload-time = "2026-11-11T06:46:08.684Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c5/71/42553557973204e0c574485304e401f54a32f39f6094439680d63e889824/dfe_schemas-0.2.29.tar.gz", hash = "sha256:00b9387d9dcc2cd6e6151612672e5c335276e99b02b521215b0f6eae79d48d8c", size = 1304657, upload-time = "2026-11-11T07:16:15.082Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/41/fb/233d56c5f08ad3ec5b233f2d775fba4b5a30531c439d7113fc0a7dd1effa/dfe_schemas-0.2.28-py3-none-any.whl", hash = "sha256:ae42d7c73df2483ee7e5d05eaeb7ffe82153b1ff27d3118a6529b6dc45f203c3", size = 2831673, upload-time = "2026-11-11T06:46:07.159Z" }, + { url = "https://files.pythonhosted.org/packages/39/cf/66a4e20949f5322e0fd299127dd1a12a3ef5fec213b4a7d5f73c4b370b84/dfe_schemas-0.2.29-py3-none-any.whl", hash = "sha256:bca1b3308c06c4aa7050455838f4bef3ba05afa7e113854705ea1679fec2e23b", size = 2837471, upload-time = "2026-11-11T07:16:13.324Z" }, ] [[package]] From a0f4c771db159eaab4b20c48965ad4309b660812 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 18:47:19 +1100 Subject: [PATCH 5/7] fix(schemas): read the header profile at its current version The default header version is no longer a constant. The external pipeline header table reads the profile at the version dfe-schemas marks current. --- src/dfe_engine/source/external.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/dfe_engine/source/external.py b/src/dfe_engine/source/external.py index 2b72775a7..0f4f9954d 100644 --- a/src/dfe_engine/source/external.py +++ b/src/dfe_engine/source/external.py @@ -32,7 +32,6 @@ from dfe_engine.schema.schema_loader import SchemaLoader from dfe_engine.source.models import ( DEFAULT_HEADER_TYPE, - DEFAULT_HEADER_VERSION, Source, external_consumer_group, transformed_topic, @@ -119,11 +118,11 @@ def reads_from_expr(field: str, expr: str) -> str | None: @functools.cache def header_fields( - header_type: str = DEFAULT_HEADER_TYPE, header_version: str = DEFAULT_HEADER_VERSION + header_type: str = DEFAULT_HEADER_TYPE, header_version: str | None = None ) -> tuple[HeaderField, ...]: """The fields the header profile reads from a record, in its column order. - Read from the pinned dfe-schemas profile, so the read order named here is the + Read from the pinned dfe-schemas profile at its current version unless one is given, so the read order named here is the one the loader applies. ``_timestamp_load``, ``_uuid``, ``_json`` and ``_raw`` carry no ``@source`` expression: the loader and ClickHouse always write those, and a value the pipeline sets is not read. From a3fc962f96726b2087e16c1556e7b2627d1373ba Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 19:05:01 +1100 Subject: [PATCH 6/7] fix(sources): report each column's cardinality The source columns response carried attributes but no cardinality, so the console could not show a saved column as `low`. The response now reports the declared cardinality. A column with the retired `lowcardinality` attribute reads as `low`. --- openapi/openapi.json | 17 +++++++++++++++++ src/dfe_engine/api/v1/sources.py | 15 +++++++++++++++ tests/unit/test_api/test_schemas_router.py | 18 ++++++++++++++++++ 3 files changed, 50 insertions(+) diff --git a/openapi/openapi.json b/openapi/openapi.json index 17a06cc39..8984cd4ea 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -56949,6 +56949,23 @@ ], "title": "Attribute" }, + "cardinality": { + "anyOf": [ + { + "type": "string", + "enum": [ + "low", + "high", + "unknown" + ] + }, + { + "type": "null" + } + ], + "title": "Cardinality", + "description": "How many distinct values the column holds, as declared. A column holding the retired lowcardinality attribute reads as low. Absent means unknown." + }, "description": { "type": "string", "title": "Description", diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py index d4340bf0d..ce3fe383f 100644 --- a/src/dfe_engine/api/v1/sources.py +++ b/src/dfe_engine/api/v1/sources.py @@ -164,6 +164,7 @@ SourceRuleError, SourceValidationError, ) +from dfe_engine.source.type_registry import DEFAULT_CARDINALITY, fold_cardinality from dfe_engine.transport import SourceTransport router = APIRouter(prefix="/sources", tags=["Sources"], dependencies=[WRITE_TURN]) @@ -944,6 +945,10 @@ class SchemaColumn(BaseModel): type: str use_case: str = "" attribute: list[str] | None = None + cardinality: Literal["low", "high", "unknown"] | None = Field( + default=None, + description="How many distinct values the column holds, as declared. A column holding the retired lowcardinality attribute reads as low. Absent means unknown.", + ) description: str = "" @@ -3871,11 +3876,21 @@ def _normalize_attribute(value: Any) -> list[str] | None: return list(value) +def _declared_cardinality(col: Any) -> str | None: + """The cardinality *col* declares, with the retired attribute read as the value it names.""" + declared = getattr(col, "cardinality", None) + if declared: + return declared + folded = fold_cardinality(None, getattr(col, "attribute", None)) + return folded if folded != DEFAULT_CARDINALITY else None + + def _to_api_schema_column(col: Any) -> SchemaColumn: return SchemaColumn( name=col.name, type=col.type, use_case=getattr(col, "use_case", "") or "", attribute=_normalize_attribute(getattr(col, "attribute", None)), + cardinality=_declared_cardinality(col), description=getattr(col, "comment", None) or getattr(col, "description", "") or "", ) diff --git a/tests/unit/test_api/test_schemas_router.py b/tests/unit/test_api/test_schemas_router.py index 272c4a0f9..b5dc8fabc 100644 --- a/tests/unit/test_api/test_schemas_router.py +++ b/tests/unit/test_api/test_schemas_router.py @@ -15,6 +15,9 @@ from types import SimpleNamespace +import pytest + +from dfe_engine.api.v1.sources import _declared_cardinality from dfe_engine.source.type_registry import TypeRegistry from tests.support.accounts import OnboardedClient, session_token from tests.support.present import present @@ -1485,3 +1488,18 @@ def test_delete_meta_schema_version(self, tmp_path): assert core_del.json()["detail"] == "Core resources can't be mutated" finally: _registries.clear() + + +@pytest.mark.parametrize( + ("column", "expected"), + [ + (SimpleNamespace(cardinality="high", attribute=["lowcardinality"]), "high"), + (SimpleNamespace(cardinality="low", attribute=[]), "low"), + (SimpleNamespace(cardinality=None, attribute=["lowcardinality", "not_null"]), "low"), + (SimpleNamespace(cardinality=None, attribute=["not_null"]), None), + (SimpleNamespace(cardinality=None, attribute=None), None), + (SimpleNamespace(attribute=None), None), + ], +) +def test_a_source_column_reports_its_declared_cardinality(column, expected): + assert _declared_cardinality(column) == expected From ce9817083e90cfdf1dd5a132459c002719544f01 Mon Sep 17 00:00:00 2001 From: Derek Date: Wed, 11 Nov 2026 19:21:39 +1100 Subject: [PATCH 7/7] fix(sources): type the column cardinality and refresh the notices The cardinality helper returns the Cardinality literal the response field declares, which ty needs. THIRD-PARTY-NOTICES lists dfe-schemas at the version the lock pins. --- THIRD-PARTY-NOTICES | 2 +- src/dfe_engine/api/v1/sources.py | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/THIRD-PARTY-NOTICES b/THIRD-PARTY-NOTICES index 1d0503342..f89d817a0 100644 --- a/THIRD-PARTY-NOTICES +++ b/THIRD-PARTY-NOTICES @@ -2290,7 +2290,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. ------------------------------------------------------------------------------ -dfe-schemas 0.2.27 (LicenseRef-HyperI) +dfe-schemas 0.2.29 (LicenseRef-HyperI) ------------------------------------------------------------------------------ Business Source License 1.1 diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py index ce3fe383f..3eb6db807 100644 --- a/src/dfe_engine/api/v1/sources.py +++ b/src/dfe_engine/api/v1/sources.py @@ -164,7 +164,7 @@ SourceRuleError, SourceValidationError, ) -from dfe_engine.source.type_registry import DEFAULT_CARDINALITY, fold_cardinality +from dfe_engine.source.type_registry import LOW_CARDINALITY, Cardinality, fold_cardinality from dfe_engine.transport import SourceTransport router = APIRouter(prefix="/sources", tags=["Sources"], dependencies=[WRITE_TURN]) @@ -945,7 +945,7 @@ class SchemaColumn(BaseModel): type: str use_case: str = "" attribute: list[str] | None = None - cardinality: Literal["low", "high", "unknown"] | None = Field( + cardinality: Cardinality | None = Field( default=None, description="How many distinct values the column holds, as declared. A column holding the retired lowcardinality attribute reads as low. Absent means unknown.", ) @@ -3876,13 +3876,13 @@ def _normalize_attribute(value: Any) -> list[str] | None: return list(value) -def _declared_cardinality(col: Any) -> str | None: +def _declared_cardinality(col: Any) -> Cardinality | None: """The cardinality *col* declares, with the retired attribute read as the value it names.""" declared = getattr(col, "cardinality", None) if declared: return declared folded = fold_cardinality(None, getattr(col, "attribute", None)) - return folded if folded != DEFAULT_CARDINALITY else None + return "low" if folded == LOW_CARDINALITY else None def _to_api_schema_column(col: Any) -> SchemaColumn: