diff --git a/THIRD-PARTY-NOTICES b/THIRD-PARTY-NOTICES index 1d050334..f89d817a 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/docs/data-plane/external-pipelines.md b/docs/data-plane/external-pipelines.md index 75e9c013..c503cfad 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 32c34d05..11614171 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, 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`. + --- ## Cardinality Axis diff --git a/docs/reference/api-route-map.md b/docs/reference/api-route-map.md index e0db7cf8..5b59d394 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 5e2770b2..8984cd4e 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -18713,6 +18713,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": [ @@ -42158,6 +42230,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", @@ -42165,7 +42242,8 @@ "effective", "stored", "origin", - "deployment_default" + "deployment_default", + "maximum" ], "title": "IntDefault", "description": "One integer default: the value in force, and where it came from." @@ -50375,6 +50453,51 @@ "title": "SchemaVersionGet", "description": "The selected version's metadata, its columns paged in the answer's ``items``." }, + "SchemaVocabulary": { + "properties": { + "cardinality": { + "$ref": "#/components/schemas/VocabularyCardinality" + }, + "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": [ + "cardinality", + "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": { @@ -56001,6 +56124,231 @@ "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." + }, + "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": { + "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": [ @@ -56601,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/scripts/route-plan.toml b/scripts/route-plan.toml index 3f811c27..c27aee0b 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 665a622a..484b38a2 100644 --- a/src/dfe_engine/api/v1/schemas.py +++ b/src/dfe_engine/api/v1/schemas.py @@ -99,6 +99,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"]) @@ -577,6 +578,106 @@ 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 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] + 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( + { + "cardinality": vocabulary.cardinality, + "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 dfa54b05..3eb6db80 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 @@ -164,6 +164,7 @@ SourceRuleError, SourceValidationError, ) +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]) @@ -944,6 +945,10 @@ class SchemaColumn(BaseModel): type: str use_case: str = "" attribute: list[str] | None = None + 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.", + ) description: str = "" @@ -3790,7 +3795,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), ) @@ -3871,11 +3876,21 @@ def _normalize_attribute(value: Any) -> list[str] | None: return list(value) +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 "low" if folded == LOW_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/src/dfe_engine/api/v1/system.py b/src/dfe_engine/api/v1/system.py index 2d8ddd34..fe46c4b3 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 84e697c5..0f4f9954 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,19 @@ 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, + Source, + external_consumer_group, + transformed_topic, +) if TYPE_CHECKING: from dfe_engine.settings import DFESettings @@ -44,43 +52,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 | None = None +) -> tuple[HeaderField, ...]: + """The fields the header profile reads from a record, in its column order. + + 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. + """ + 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 +202,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 c6695681..c0275c1d 100644 --- a/src/dfe_engine/source/type_registry.py +++ b/src/dfe_engine/source/type_registry.py @@ -126,6 +126,93 @@ 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 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, ...] + 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 +482,110 @@ 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) + 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, + 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/src/vulture_whitelist.py b/src/vulture_whitelist.py index 8dbf8ec0..9ffbe215 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 diff --git a/tests/unit/test_api/test_schemas_router.py b/tests/unit/test_api/test_schemas_router.py index eab9946b..b5dc8fab 100644 --- a/tests/unit/test_api/test_schemas_router.py +++ b/tests/unit/test_api/test_schemas_router.py @@ -15,6 +15,10 @@ 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 @@ -23,6 +27,54 @@ 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_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() + + 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).""" @@ -1436,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 diff --git a/tests/unit/test_api/test_system_defaults.py b/tests/unit/test_api/test_system_defaults.py index 7274b91b..59b87adf 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 1346ff4c..7da29ae6 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 f90996f7..761ff61f 100644 --- a/tests/unit/test_source/test_type_registry.py +++ b/tests/unit/test_source/test_type_registry.py @@ -397,3 +397,163 @@ 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"]}, + }, + "cardinality": { + "values": ["low", "high", "unknown"], + "default": "unknown", + "labels": {"low": "Few values"}, + }, + "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_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} + + 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): + 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() + + assert len(vocabulary.primitives) == len(registry.primitives) + assert all( + entry.label + for entry in (*vocabulary.primitives, *vocabulary.use_cases, *vocabulary.attributes) + )