diff --git a/.github/workflows/flink-iceberg.yml b/.github/workflows/flink-iceberg.yml index a0eb3ed7..735d686c 100644 --- a/.github/workflows/flink-iceberg.yml +++ b/.github/workflows/flink-iceberg.yml @@ -17,7 +17,7 @@ on: - "fluid_build/providers/_iceberg*.py" - "fluid_build/iac/providers/aws.py" - "fluid_build/build_runners/base.py" - - "fluid_build/api/changelog_rules.py" + - "fluid_build/api/changelog*.py" - "examples/flink-iceberg/**" - "tests/streaming/**" - "tests/integration/test_flink_iceberg_live.py" diff --git a/.gitignore b/.gitignore index 0bd8a106..cdb04302 100644 --- a/.gitignore +++ b/.gitignore @@ -175,6 +175,8 @@ dmypy.json /*-fluid.yaml /*.fluid.yaml fluid.workspace.yaml +# An example that shares fragments across contracts needs its workspace file. +!/examples/flink-iceberg/shapes/fluid.workspace.yaml # --- fluid-cli: engineer-personal state (never commit) ------------------- # Receipts, logs, and per-engineer learning history live under .fluid/ diff --git a/examples/flink-iceberg/README.md b/examples/flink-iceberg/README.md index b3b2f3e8..29ad5b3c 100644 --- a/examples/flink-iceberg/README.md +++ b/examples/flink-iceberg/README.md @@ -6,6 +6,55 @@ operator prerequisites, workflow, limits and certification status. The [contract](contract.fluid.yaml) preserves five nested structs, arrays of structs, exact decimal prices, maps and event-time timestamps. +## Record shapes + +A build declares the records it reads in `properties.stream`: +- the topic; +- the value's format (`json` or `avro-confluent`) and its fields, as columns; +- optionally the record key (`raw`, `json` or `avro-confluent`); +- optionally Kafka metadata (`topic`, `partition`, `offset`, `timestamp`, + `timestamp-type`, `headers`, `leader-epoch`), each read as a column. + +`properties.mapping` says how the records become the output's rows. forge-cli +writes the query from the two, and checks every output column's type and +nullability against the path it is read from before anything deploys. An output +column the mapping does not list is read from the stream column at the same +path. So a stream landed as it is, flat or nested, needs no mapping, like the +contract above. + +The [shapes](shapes) directory reads one stream, kept in a +[fragment](shapes/fragments/streams/orders.yaml), into two products: +- [order-items](shapes/order-items/contract.fluid.yaml) flattens each event to + one row per item. It lifts nested fields to columns, explodes `items`, looks + up a map entry and a header, and keeps the record key and the Kafka position. +- [orders-nested](shapes/orders-nested/contract.fluid.yaml) writes each event as + a nested row in a shape of its own: + - structs built from fields that are flat or elsewhere in the record (a + dotted name such as `order.key` maps one field); + - a subset of `customer` that stays null where the customer is; + - the coordinates struct moved up; + - the Kafka position grouped in one struct. + +Their `$ref`s reach the shared fragment through the +[workspace](shapes/fluid.workspace.yaml). Check a fragment on its own, naming +the schema version because the workspace has no root contract: + +```bash +fluid validate --schema-version 0.7.7 --fragment streamSource shapes/fragments/streams/orders.yaml +``` + +`--fragment recordMapping` checks a mapping, and `--fragment schema` a column list. + +A build may run its own `properties.sql` instead, read by output column name. +That SQL runs with the job's catalog identity, so it can read any table that +identity can read, not only the stream. `fluid apply` therefore refuses it unless +the operator lists it in `FLUID_FLINK_CONTRACT_SQL_ALLOWLIST`. + +List it as `/@sha256:`, which the refusal prints. That +entry approves the SQL that was reviewed and nothing it is later edited into. +`/` and `` approve whatever SQL the build or product +runs. + ## Run the data-plane proof locally Requirements: a working Docker daemon with Compose, Forge's test environment, @@ -26,9 +75,11 @@ normal `/tmp` or hidden home directories. The test uses a unique Compose project and loopback-only random host ports, produces events, restarts the task manager, produces another batch, independently -compares every nested value and then checks schema-drift refusal. Containers, -networks and volumes are removed in `finally`. Fixture credentials are local test -values only. Docker images and build cache are retained. +compares every nested value and then checks schema-drift refusal. A second test +runs both shapes from the same keyed records with a header, and compares their +rows with what was produced. Containers, networks and volumes are removed in +`finally`. Fixture credentials are local test values only. Docker images and +build cache are retained. Do not run `compose up` directly without generating/mounting `job.properties`; the test does that from the actual compiler output. diff --git a/examples/flink-iceberg/compose.yaml b/examples/flink-iceberg/compose.yaml index b1b886f0..b5169957 100644 --- a/examples/flink-iceberg/compose.yaml +++ b/examples/flink-iceberg/compose.yaml @@ -18,33 +18,24 @@ services: - 0.0.0.0:9092 - --advertise-kafka-addr - kafka:9092 - minio: - image: minio/minio:RELEASE.2025-04-22T22-12-26Z - command: server /data + s3: + # MinIO no longer publishes its images (Docker Hub and Quay both refuse the pinned + # tags), so the fixture's object store is Adobe's S3Mock (Apache-2.0). It creates the + # warehouse bucket at startup, accepts any credentials and serves path-style + # requests only, which is how every client here is configured. + image: adobe/s3mock:5.2.3 environment: - MINIO_ROOT_USER: minioadmin - MINIO_ROOT_PASSWORD: minioadmin # pragma: allowlist secret - disposable local fixture - init-storage: - image: minio/mc:RELEASE.2025-04-16T18-13-26Z - depends_on: [minio] - entrypoint: - - /bin/sh - - -ec - - | - until mc alias set test http://minio:9000 minioadmin minioadmin; do sleep 1; done - mc mb --ignore-existing test/warehouse + COM_ADOBE_TESTING_S3MOCK_STORE_INITIAL_BUCKETS: warehouse iceberg-rest: image: tabulario/iceberg-rest:1.6.0 - depends_on: - init-storage: - condition: service_completed_successfully + depends_on: [s3] environment: AWS_ACCESS_KEY_ID: minioadmin AWS_SECRET_ACCESS_KEY: minioadmin # pragma: allowlist secret - disposable local fixture AWS_REGION: us-east-1 CATALOG_WAREHOUSE: s3://warehouse/ CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO - CATALOG_S3_ENDPOINT: http://minio:9000 + CATALOG_S3_ENDPOINT: http://s3:9090 CATALOG_S3_PATH__STYLE__ACCESS: "true" ports: ["127.0.0.1:0:8181"] jobmanager: @@ -57,6 +48,7 @@ services: FLINK_PROPERTIES: | jobmanager.rpc.address: jobmanager parallelism.default: 1 + taskmanager.numberOfTaskSlots: 2 execution.checkpointing.interval: 2s execution.checkpointing.mode: EXACTLY_ONCE state.checkpoints.dir: file:///tmp/forge-checkpoints diff --git a/examples/flink-iceberg/contract.fluid.yaml b/examples/flink-iceberg/contract.fluid.yaml index 59a4f740..f10625de 100644 --- a/examples/flink-iceberg/contract.fluid.yaml +++ b/examples/flink-iceberg/contract.fluid.yaml @@ -17,64 +17,66 @@ builds: platform: kubernetes properties: language: flink_sql - sql: >- - SELECT event_id, customer, items, attributes, event_time FROM source_events - parameters: - flink: - management: operator - source: - bootstrapServers: kafka:9092 - topic: orders - groupId: forge-commerce-orders-v1 - format: json - eventTime: - column: event_time - delaySeconds: 5 - schema: - - name: event_id + # The records, as the topic carries them. With no mapping, each output + # column is read from the stream column of the same name, nested as it is. + stream: + kind: kafka + topic: orders + format: json + schema: + - name: event_id + type: string + required: true + - name: customer + type: struct + fields: + - name: id type: string - required: true - - name: customer + - name: address type: struct fields: - - name: id + - name: country type: string - - name: address + - name: geo type: struct fields: - - name: country - type: string - - name: geo + - name: location type: struct fields: - - name: location + - name: coordinates type: struct fields: - - name: coordinates - type: struct - fields: - - name: latitude - type: double - - name: longitude - type: double - - name: items - type: array - element: - type: struct - fields: - - name: sku - type: string - - name: price - type: decimal(18,2) - - name: attributes - type: map - key: - type: string - required: true - value: + - name: latitude + type: double + - name: longitude + type: double + - name: items + type: array + element: + type: struct + fields: + - name: sku type: string - - name: event_time - type: timestamp_ltz(3) + - name: price + type: decimal(18,2) + - name: attributes + type: map + key: + type: string + required: true + value: + type: string + - name: event_time + type: timestamp_ltz(3) + parameters: + flink: + management: operator + source: + bootstrapServers: kafka:9092 + groupId: forge-commerce-orders-v1 + eventTime: + column: event_time + delaySeconds: 5 deployment: name: commerce-orders namespace: data-products diff --git a/examples/flink-iceberg/shapes/fluid.workspace.yaml b/examples/flink-iceberg/shapes/fluid.workspace.yaml new file mode 100644 index 00000000..23ca221c --- /dev/null +++ b/examples/flink-iceberg/shapes/fluid.workspace.yaml @@ -0,0 +1,3 @@ +# Both contracts below read the one stream fragment in fragments/, so the +# workspace lets their $refs reach this directory. +ref_root: . diff --git a/examples/flink-iceberg/shapes/fragments/streams/orders.value.yaml b/examples/flink-iceberg/shapes/fragments/streams/orders.value.yaml new file mode 100644 index 00000000..4242958b --- /dev/null +++ b/examples/flink-iceberg/shapes/fragments/streams/orders.value.yaml @@ -0,0 +1,45 @@ +# The value of each record on the orders topic, as the producers write it. +- name: event_id + type: string + required: true +- name: customer + type: struct + fields: + - name: id + type: string + - name: address + type: struct + fields: + - name: country + type: string + - name: geo + type: struct + fields: + - name: location + type: struct + fields: + - name: coordinates + type: struct + fields: + - name: latitude + type: double + - name: longitude + type: double +- name: items + type: array + element: + type: struct + fields: + - name: sku + type: string + - name: price + type: decimal(18,2) +- name: attributes + type: map + key: + type: string + required: true + value: + type: string +- name: event_time + type: timestamp_ltz(3) diff --git a/examples/flink-iceberg/shapes/fragments/streams/orders.yaml b/examples/flink-iceberg/shapes/fragments/streams/orders.yaml new file mode 100644 index 00000000..e76de651 --- /dev/null +++ b/examples/flink-iceberg/shapes/fragments/streams/orders.yaml @@ -0,0 +1,21 @@ +# The orders topic, shared by every product that reads it: the record value, +# the key and the Kafka metadata each product may read as columns. +kind: kafka +topic: orders +format: json +schema: + $ref: ./orders.value.yaml +key: + format: raw + schema: + - name: order_key + type: string +metadata: + - name: kafka_partition + from: partition + - name: kafka_offset + from: offset + - name: kafka_timestamp + from: timestamp + - name: kafka_headers + from: headers diff --git a/examples/flink-iceberg/shapes/order-items/contract.fluid.yaml b/examples/flink-iceberg/shapes/order-items/contract.fluid.yaml new file mode 100644 index 00000000..7166b5c4 --- /dev/null +++ b/examples/flink-iceberg/shapes/order-items/contract.fluid.yaml @@ -0,0 +1,117 @@ +fluidVersion: "0.7.7" +kind: DataProduct +id: commerce.order-items +name: Order items +description: >- + One row per item of each order event: nested fields lifted to columns, the + items array exploded, a map entry and a header looked up, and the record + key and Kafka position kept for lineage. +metadata: + owner: + team: data-platform +builds: + - id: ingest-order-items + engine: flink + pattern: embedded-logic + outputs: [order_items] + execution: + trigger: + type: continuous + runtime: + platform: kubernetes + properties: + language: flink_sql + stream: + $ref: ../fragments/streams/orders.yaml + mapping: + # One output row per element of items; an order with no items writes none. + explode: + path: items + as: item + # Only the columns read from another path are listed: event_id, + # order_key, kafka_partition, kafka_offset and event_time come by name. + columns: + - name: customer_id + from: customer.id + - name: country + from: customer.address.country + - name: latitude + from: customer.address.geo.location.coordinates.latitude + - name: longitude + from: customer.address.geo.location.coordinates.longitude + - name: channel + from: attributes['channel'] + - name: sku + from: item.sku + - name: price + from: item.price + - name: trace_id + from: kafka_headers['trace-id'] + cast: string + parameters: + flink: + management: operator + source: + bootstrapServers: kafka:9092 + groupId: forge-commerce-order-items-v1 + eventTime: + column: event_time + delaySeconds: 5 + deployment: + name: commerce-order-items + namespace: data-products + context: development + serviceAccount: flink + # Replace with the digest of the image built from ../../runtime/Dockerfile. + image: forge-flink@sha256:0000000000000000000000000000000000000000000000000000000000000000 + checkpointUri: s3://flink-state/order-items/checkpoints + savepointUri: s3://flink-state/order-items/savepoints + haUri: s3://flink-state/order-items/ha + secretRefs: [flink-storage] + maxSnapshotAgeSeconds: 300 +exposes: + - exposeId: order_items + kind: table + binding: + platform: aws + format: iceberg + location: + catalog: rest + uri: http://iceberg-rest:8181 + warehouse: s3://warehouse/ + database: commerce + table: order_items + icebergConfig: + writeVersion: 2 + fileFormat: parquet + contract: + schema: + - name: event_id + type: string + required: true + - name: order_key + type: string + - name: customer_id + type: string + - name: country + type: string + - name: latitude + type: double + - name: longitude + type: double + - name: channel + type: string + - name: sku + type: string + - name: price + type: decimal(18,2) + - name: trace_id + type: string + - name: kafka_partition + type: int + required: true + - name: kafka_offset + type: long + required: true + - name: event_time + type: timestamp_ltz(3) diff --git a/examples/flink-iceberg/shapes/orders-nested/contract.fluid.yaml b/examples/flink-iceberg/shapes/orders-nested/contract.fluid.yaml new file mode 100644 index 00000000..e53a2580 --- /dev/null +++ b/examples/flink-iceberg/shapes/orders-nested/contract.fluid.yaml @@ -0,0 +1,136 @@ +fluidVersion: "0.7.7" +kind: DataProduct +id: commerce.orders-nested +name: Orders, nested +description: >- + Each order event written as a nested Iceberg row in a shape of its own: + structs built from fields that are flat or elsewhere in the record, a + subset of the customer that stays null where the customer is, the + coordinates struct moved up, the items kept as they are, and the Kafka + position grouped in one struct. +metadata: + owner: + team: data-platform +builds: + - id: ingest-orders-nested + engine: flink + pattern: embedded-logic + outputs: [orders_nested] + execution: + trigger: + type: continuous + runtime: + platform: kubernetes + properties: + language: flink_sql + stream: + $ref: ../fragments/streams/orders.yaml + mapping: + # A dotted name maps one field of a struct output; the struct is built + # from its fields. Fields it does not list come from the same path. + columns: + - name: order.key + from: order_key + - name: order.channel + from: attributes['channel'] + - name: order.placed_at + from: event_time + - name: customer.country + from: customer.address.country + - name: location + from: customer.address.geo.location.coordinates + - name: kafka.partition + from: kafka_partition + - name: kafka.offset + from: kafka_offset + - name: kafka.timestamp + from: kafka_timestamp + parameters: + flink: + management: operator + source: + bootstrapServers: kafka:9092 + groupId: forge-commerce-orders-nested-v1 + eventTime: + column: event_time + delaySeconds: 5 + deployment: + name: commerce-orders-nested + namespace: data-products + context: development + serviceAccount: flink + # Replace with the digest of the image built from ../../runtime/Dockerfile. + image: forge-flink@sha256:0000000000000000000000000000000000000000000000000000000000000000 + checkpointUri: s3://flink-state/orders-nested/checkpoints + savepointUri: s3://flink-state/orders-nested/savepoints + haUri: s3://flink-state/orders-nested/ha + secretRefs: [flink-storage] + maxSnapshotAgeSeconds: 300 +exposes: + - exposeId: orders_nested + kind: table + binding: + platform: aws + format: iceberg + location: + catalog: rest + uri: http://iceberg-rest:8181 + warehouse: s3://warehouse/ + database: commerce + table: orders_nested + icebergConfig: + writeVersion: 2 + fileFormat: parquet + contract: + schema: + - name: event_id + type: string + required: true + - name: order + type: struct + required: true + fields: + - name: key + type: string + - name: channel + type: string + - name: placed_at + type: timestamp_ltz(3) + - name: customer + type: struct + fields: + - name: id + type: string + - name: country + type: string + - name: location + type: struct + fields: + - name: latitude + type: double + - name: longitude + type: double + - name: items + type: array + element: + type: struct + fields: + - name: sku + type: string + - name: price + type: decimal(18,2) + - name: kafka + type: struct + required: true + fields: + - name: partition + type: int + required: true + - name: offset + type: long + required: true + - name: timestamp + type: timestamp_ltz(3) + required: true + - name: event_time + type: timestamp_ltz(3) diff --git a/fluid_build/api/changelog.py b/fluid_build/api/changelog.py index 767653ed..3c0efdc8 100644 --- a/fluid_build/api/changelog.py +++ b/fluid_build/api/changelog.py @@ -44,6 +44,7 @@ diff_sovereignty, iter_expose_pairs, ) +from .changelog_stream_rules import diff_streams from .changelog_types import Change, ChangelogReport @@ -98,6 +99,9 @@ def compare_contracts( report.extend(diff_agent_policy(baseline, new)) report.extend(diff_sovereignty(baseline, new)) + # What streaming builds read (builds[].properties.stream). + report.extend(diff_streams(baseline, new)) + # Per-expose schema and quality. for expose_id, idx, base_expose, new_expose in iter_expose_pairs(baseline, new): if base_expose is None and new_expose is not None: diff --git a/fluid_build/api/changelog_stream_rules.py b/fluid_build/api/changelog_stream_rules.py new file mode 100644 index 00000000..4eb34c06 --- /dev/null +++ b/fluid_build/api/changelog_stream_rules.py @@ -0,0 +1,316 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""``fluid diff`` rules for what a streaming build reads: ``builds[].properties.stream``. + +A stream's schema is a reader schema: it declares what the job reads, and producers +evolve ahead of it. The column rules apply to it as they are, with one exception. A +field the job stops reading is not breaking, because an output that still reads it +fails ``fluid validate``. A field it newly requires is breaking, because a record +without it stops the job: Confluent Schema Registry's FORWARD compatibility, seen +from the consumer. What the outputs then contain is diffed with the exposes. + +A different topic or encoding is breaking. So is an explode added, removed or moved, +because it changes how many rows each record becomes, and so is a build that starts +running, or changes, SQL written in the contract, which can read any table the job's +catalog identity can. An output read from another +path keeps its type but may change meaning; that is reported for review, as dbt +reports a changed model, not as breaking. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Mapping, Optional, Tuple + +from .changelog_rules import _diff_nested_fields +from .changelog_types import Change + + +def _builds(contract: Mapping[str, Any]) -> Dict[str, Tuple[int, Mapping[str, Any]]]: + """``{build id: (index, properties)}`` for every build with properties.""" + found: Dict[str, Tuple[int, Mapping[str, Any]]] = {} + for i, build in enumerate(contract.get("builds") or []): + props = build.get("properties") if isinstance(build, Mapping) else None + if isinstance(props, Mapping): + found[str(build.get("id"))] = (i, props) + return found + + +def _has_stream(props: Mapping[str, Any]) -> bool: + return isinstance(props.get("stream"), Mapping) + + +def _label(relative: str) -> str: + """A schema path as a reader names it: ``customer.id``, ``items[].price``.""" + for suffix in (".type", ".required", ".nullable"): + if relative.endswith(suffix): + relative = relative[: -len(suffix)] + for raw, shown in ((".element", "[]"), (".key", "{key}"), (".value", "{value}")): + relative = relative.replace(raw, shown) + return relative.replace(".fields.", ".") + + +def _fields(stream: Mapping[str, Any], *keys: str) -> List[Mapping[str, Any]]: + node: Any = stream + for key in keys: + node = node.get(key) if isinstance(node, Mapping) else None + return [f for f in node or [] if isinstance(f, Mapping)] if isinstance(node, list) else [] + + +def _reader_changes(changes: List[Change], prefix: str) -> List[Change]: + """The column rules' findings, restated for a schema the job reads.""" + out: List[Change] = [] + for change in changes: + field = _label(change.path[len(prefix) + 1 :]) + kind = "stream_" + change.kind.removeprefix("nested_") + severity = change.severity + if change.kind == "nested_field_removed": + severity = "non_breaking" + description = f"the job no longer reads '{field}' (an output still reading it fails)" + elif change.kind == "nested_field_added": + description = ( + f"the job now reads '{field}'" + if severity == "non_breaking" + else f"the job now requires '{field}': a record without it stops the job" + ) + elif change.kind == "nested_nullable_tightened": + description = f"'{field}' is now required: a record without it stops the job" + elif change.kind == "nested_nullable_loosened": + description = f"'{field}' may now be null" + else: + description = f"'{field}' type changed {change.before} -> {change.after}" + out.append( + Change( + path=change.path, + kind=kind, + severity=severity, + description=description, + before=change.before, + after=change.after, + ) + ) + return out + + +def _sources(props: Mapping[str, Any]) -> Dict[str, str]: + """``{output: source}``, a conversion included: ``kafka_headers['trace-id'] as string``.""" + mapping = props.get("mapping") + columns = mapping.get("columns") if isinstance(mapping, Mapping) else None + return { + str(entry.get("name")): str(entry.get("from")) + + (f" as {entry['cast']}" if entry.get("cast") is not None else "") + for entry in columns or [] + if isinstance(entry, Mapping) + } + + +def _sql(props: Mapping[str, Any]) -> Optional[str]: + sql = props.get("sql") + return sql.strip() if isinstance(sql, str) else None + + +def _explode(props: Mapping[str, Any]) -> Optional[str]: + mapping = props.get("mapping") + explode = mapping.get("explode") if isinstance(mapping, Mapping) else None + return str(explode.get("path")) if isinstance(explode, Mapping) else None + + +def _sql_change( + build_id: str, index: int, base_props: Mapping[str, Any], new_props: Mapping[str, Any] +) -> Optional[Change]: + """A build's own SQL added, changed or removed: what it reads is the reviewer's call.""" + old_sql, new_sql = _sql(base_props), _sql(new_props) + if old_sql == new_sql: + return None + if old_sql is None: + kind, severity, text = ( + "stream_contract_sql_added", + "breaking", + f"build '{build_id}' runs SQL written in the contract: it can read any table the " + "job's catalog identity can", + ) + elif new_sql is None: + kind, severity, text = ( + "stream_contract_sql_removed", + "non_breaking", + f"build '{build_id}' reads the stream through its mapping instead of its own SQL", + ) + else: + kind, severity, text = ( + "stream_contract_sql_changed", + "breaking", + f"build '{build_id}' runs different SQL written in the contract: review what it " + "now reads", + ) + return Change( + path=f"builds[{index}].properties.sql", + kind=kind, + severity=severity, + description=text, + before=old_sql, + after=new_sql, + ) + + +def diff_streams(baseline: Mapping[str, Any], new: Mapping[str, Any]) -> List[Change]: + """Classify changes to the stream every build present in both contracts reads.""" + base_builds = _builds(baseline) + new_builds = _builds(new) + changes: List[Change] = [] + # A build the baseline does not have (or had under another id) that runs its own + # SQL over a stream: what it reads is the reviewer's to check. + for build_id in sorted(new_builds.keys() - base_builds.keys()): + index, props = new_builds[build_id] + if _has_stream(props) and _sql(props) is not None: + changes.append( + Change( + path=f"builds[{index}].properties.sql", + kind="stream_contract_sql_added", + severity="breaking", + description=( + f"new build '{build_id}' runs SQL written in the contract: it can read " + "any table the job's catalog identity can" + ), + before=None, + after=_sql(props), + ) + ) + for build_id in sorted(base_builds.keys() & new_builds.keys()): + _, base_props = base_builds[build_id] + index, new_props = new_builds[build_id] + if not (_has_stream(base_props) or _has_stream(new_props)): + continue + # Before anything else: a build may gain its stream and new SQL in one edit. + sql_change = _sql_change(build_id, index, base_props, new_props) + if sql_change is not None: + changes.append(sql_change) + if _has_stream(base_props) != _has_stream(new_props): + declared = _has_stream(new_props) + changes.append( + Change( + path=f"builds[{index}].properties.stream", + kind="stream_declared" if declared else "stream_dropped", + severity="info", + description=( + f"build '{build_id}' " + + ( + "now declares the stream it reads" + if declared + else "no longer declares its stream" + ) + ), + before=None if declared else dict(base_props["stream"]), + after=dict(new_props["stream"]) if declared else None, + ) + ) + continue + old, cur = base_props["stream"], new_props["stream"] + at = f"builds[{index}].properties" + + def changed( + field: str, kind: str, severity: str, text: str, before: Any, after: Any + ) -> None: + changes.append( + Change( + path=f"{at}.{field}", + kind=kind, + severity=severity, + description=text, + before=before, + after=after, + ) + ) + + for key, what in (("topic", "topic"), ("format", "value format")): + if old.get(key) != cur.get(key): + changed( + f"stream.{key}", + f"stream_{key}_changed", + "breaking", + f"build '{build_id}' reads another {what}: " + f"{old.get(key)!r} -> {cur.get(key)!r}", + old.get(key), + cur.get(key), + ) + old_key, new_key = old.get("key"), cur.get("key") + old_key_format = old_key.get("format") if isinstance(old_key, Mapping) else None + new_key_format = new_key.get("format") if isinstance(new_key, Mapping) else None + if old_key_format and new_key_format and old_key_format != new_key_format: + changed( + "stream.key.format", + "stream_key_format_changed", + "breaking", + f"build '{build_id}' decodes the record key as {new_key_format!r}, " + f"not {old_key_format!r}", + old_key_format, + new_key_format, + ) + for keys in (("schema",), ("key", "schema")): + prefix = f"{at}.stream.{'.'.join(keys)}" + changes.extend( + _reader_changes( + _diff_nested_fields( + _fields(old, *keys), + _fields(cur, *keys), + expose_id=f"stream of build {build_id}", + path_prefix=prefix, + ), + prefix, + ) + ) + old_meta = { + str(m.get("name")): m.get("from") + for m in old.get("metadata") or [] + if isinstance(m, Mapping) + } + new_meta = { + str(m.get("name")): m.get("from") + for m in cur.get("metadata") or [] + if isinstance(m, Mapping) + } + for name in sorted(old_meta.keys() & new_meta.keys()): + if old_meta[name] != new_meta[name]: + changed( + "stream.metadata", + "stream_metadata_source_changed", + "breaking", + f"metadata column '{name}' reads {new_meta[name]!r}, not {old_meta[name]!r}", + old_meta[name], + new_meta[name], + ) + if _explode(base_props) != _explode(new_props): + changed( + "mapping.explode", + "stream_explode_changed", + "breaking", + f"build '{build_id}' turns each record into rows of " + f"{_explode(new_props) or 'one'} instead of {_explode(base_props) or 'one'}", + _explode(base_props), + _explode(new_props), + ) + old_sources, new_sources = _sources(base_props), _sources(new_props) + for name in sorted(old_sources.keys() | new_sources.keys()): + before = old_sources.get(name, name) + after = new_sources.get(name, name) + if before != after: + changed( + "mapping.columns", + "stream_column_source_changed", + "non_breaking", + f"output '{name}' is read from '{after}' instead of '{before}' " + "(same type; review what it now means)", + before, + after, + ) + return changes diff --git a/fluid_build/providers/odcs/mappers/schema.py b/fluid_build/providers/odcs/mappers/schema.py index 6886aee8..72fef353 100644 --- a/fluid_build/providers/odcs/mappers/schema.py +++ b/fluid_build/providers/odcs/mappers/schema.py @@ -38,7 +38,7 @@ import copy from collections.abc import Mapping -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Sequence from .base import ( ExportCtx, @@ -217,6 +217,11 @@ def _property_to_field(prop: Mapping[str, Any]) -> Dict[str, Any]: ("criticalDataElement", "critical_data_element"), ("authoritativeDefinitions", "authoritative_definitions"), ("customProperties", "custom_properties"), + # Property lineage: FLUID derives it from a build's stream mapping on export, + # and an imported contract carries no build, so it is kept as written. + ("transformSourceObjects", "transform_source_objects"), + ("transformLogic", "transform_logic"), + ("transformDescription", "transform_description"), ): if src in prop: pt[dst] = prop[src] @@ -267,6 +272,7 @@ def _extract_schema(fluid: Mapping[str, Any], include_quality: bool = True) -> L continue expose_id = expose.get("exposeId") or expose.get("id") or "dataset" + grain = _add_stream_lineage(fluid, expose_id, contract_schema, properties) pt = get_expose_passthrough(expose) physical_type = pt.get("physical_type") or _platform_to_physical_type(provider) @@ -285,6 +291,8 @@ def _extract_schema(fluid: Mapping[str, Any], include_quality: bool = True) -> L schema_object["physicalName"] = physical_name if expose.get("description"): schema_object["description"] = expose["description"] + if grain: + schema_object["dataGranularityDescription"] = grain # Pass-through extras for src, dst in ( @@ -303,6 +311,91 @@ def _extract_schema(fluid: Mapping[str, Any], include_quality: bool = True) -> L return odcs_schema +def _add_stream_lineage( + fluid: Mapping[str, Any], + expose_id: Any, + columns: List[Mapping[str, Any]], + properties: List[Dict[str, Any]], +) -> Optional[str]: + """Mark each property read from a build's ``properties.stream`` with its source. + + ODCS v3's property lineage: the topic as ``transformSourceObjects`` and the path the + column (or a nested field mapped on its own) is read from as ``transformLogic``. + Returns the row grain an explode sets, for ``dataGranularityDescription``. + """ + builds = [ + build + for build in fluid.get("builds") or [] + if isinstance(build, Mapping) + and expose_id in (build.get("outputs") or []) + and isinstance(build.get("properties"), Mapping) + ] + if len(builds) != 1: + return None + lineage = _stream_lineage(builds[0]["properties"], columns) + if lineage is None: + return None + topic, paths, grain = lineage + + def mark(prefix: str, props: List[Dict[str, Any]]) -> None: + for prop in props: + name = f"{prefix}{prop.get('name')}" + if name in paths: + prop["transformSourceObjects"] = [topic] + prop["transformLogic"] = paths[name] + elif isinstance(prop.get("properties"), list): + mark(name + ".", prop["properties"]) + + mark("", properties) + return grain + + +def _stream_lineage( + properties: Mapping[str, Any], target: Sequence[Any] +) -> tuple[str, dict[str, str], str | None] | None: + """Where each output column, and each nested field mapped on its own, is read from. + + For catalogs (ODCS ``transformSourceObjects``/``transformLogic``): the stream's topic, + ``{output path: source path}`` and the row grain an explode sets. It describes the + contract and does not compile it, so `fluid validate` still reports a mapping that + cannot be compiled; a build that runs its own SQL has no lineage to report. + """ + stream = properties.get("stream") + mapping = properties.get("mapping") or {} + if properties.get("sql") is not None or not isinstance(stream, Mapping): + return None + topic = stream.get("topic") + if not isinstance(topic, str) or not isinstance(mapping, Mapping): + return None + overrides = { + entry["name"]: entry["from"] + for entry in mapping.get("columns") or [] + if isinstance(entry, Mapping) + and isinstance(entry.get("name"), str) + and isinstance(entry.get("from"), str) + } + paths: dict[str, str] = {} + + def walk(prefix: str, fields: Any) -> None: + for field in fields if isinstance(fields, list) else []: + if not isinstance(field, Mapping) or not isinstance(field.get("name"), str): + continue + name = prefix + field["name"] + if name in overrides: + paths[name] = overrides[name] + elif any(key.startswith(name + ".") for key in overrides): + walk(name + ".", field.get("fields")) + else: + paths[name] = name + + walk("", target) + explode = mapping.get("explode") + grain = None + if isinstance(explode, Mapping) and isinstance(explode.get("path"), str): + grain = f"One row per element of {explode['path']} in each record of topic {topic}" + return topic, paths, grain + + def _expose_provider(expose: Mapping[str, Any]) -> Optional[str]: binding = expose.get("binding") if isinstance(binding, Mapping): @@ -434,6 +527,9 @@ def _field_to_property( ("critical_data_element", "criticalDataElement"), ("authoritative_definitions", "authoritativeDefinitions"), ("custom_properties", "customProperties"), + ("transform_source_objects", "transformSourceObjects"), + ("transform_logic", "transformLogic"), + ("transform_description", "transformDescription"), ): if src in pt: prop[dst] = pt[src] diff --git a/fluid_build/schemas/fluid-schema-0.7.7.json b/fluid_build/schemas/fluid-schema-0.7.7.json index 61b6b5aa..393a2b99 100644 --- a/fluid_build/schemas/fluid-schema-0.7.7.json +++ b/fluid_build/schemas/fluid-schema-0.7.7.json @@ -1187,6 +1187,136 @@ "value": {"$ref": "#/$defs/nestedType"} } }, + "streamSource": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): The Kafka records a streaming build reads, described apart from the engine that reads them: one topic, how its values are encoded, the value's fields, an optional key, and Kafka record metadata read as columns. The value and key schemas are lists of columns, the same shape as an expose's contract.schema, so one fragment can describe a topic and a table that lands it. Brokers, consumer groups and registries are the engine's parameters, not part of the stream.", + "required": ["kind", "topic", "format", "schema"], + "properties": { + "kind": { + "const": "kafka", + "description": "The streaming system. Only kafka in this preview." + }, + "topic": { + "type": "string", + "pattern": "^(?!\\.\\.?(?![\\s\\S]))[A-Za-z0-9._-]{1,249}(?![\\s\\S])", + "description": "One Kafka topic, by name: Kafka's legal characters (not . or ..), no pattern and no list." + }, + "format": { + "enum": ["json", "avro-confluent"], + "description": "How record values are encoded. json: one JSON object per record; a field the schema does not declare is ignored, and a declared field that is missing reads as null. avro-confluent: Avro in the Confluent Schema Registry wire format. The declared schema is the reader schema, so the producers' subject must stay readable by it: FORWARD or FULL compatibility, transitive." + }, + "schema": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/column"}, + "description": "The record value's fields, as columns. Nested data uses fields (struct), element (array) and key/value (map)." + }, + "key": {"$ref": "#/$defs/streamKey"}, + "metadata": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/streamMetadataColumn"}, + "description": "Kafka record metadata, each read as a column." + } + } + }, + "streamKey": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): The record key, read as columns next to the value's. A record without a key reads its key columns as null.", + "required": ["format", "schema"], + "properties": { + "format": { + "enum": ["raw", "json", "avro-confluent"], + "description": "raw: the key's bytes as one column, of type string (UTF-8) or binary. json and avro-confluent: a key record whose fields are the schema's columns." + }, + "schema": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/column"}, + "description": "The key's fields, as columns: exactly one for raw." + }, + "prefix": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(?![\\s\\S])", + "description": "Put in front of every key column's name, so a key field can share its name with a value field (a Debezium key repeats the value's id). A mapping names the prefixed column." + } + } + }, + "streamMetadataColumn": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): One piece of Kafka record metadata, read as a column. Its type is fixed by what it reads: topic string, partition int, offset long, timestamp timestamp_ltz(3), timestamp-type string, headers map and leader-epoch int. leader-epoch is null when the broker does not report it; the others are never null.", + "required": ["name", "from"], + "properties": { + "name": { + "type": "string", + "minLength": 1, + "description": "The column's name, unique among the stream's value, key and metadata columns." + }, + "from": { + "enum": ["topic", "partition", "offset", "timestamp", "timestamp-type", "headers", "leader-epoch"], + "description": "The metadata the column reads." + }, + "description": {"type": "string"} + } + }, + "recordMapping": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): How the stream's records become the output's rows. An output column the mapping does not list, nested fields included, is read from the stream column or field at the same path, and keeps its type: so a stream landed as it is, flat or nested, needs no mapping. forge-cli writes the query from the mapping and checks every output column's type and nullability against it before anything deploys; no contract text reaches the job's SQL.", + "properties": { + "explode": {"$ref": "#/$defs/recordExplode"}, + "columns": { + "type": "array", + "minItems": 1, + "items": {"$ref": "#/$defs/recordMappingColumn"}, + "description": "Output columns, or nested fields of them, that are read from another path, renamed or converted." + } + } + }, + "recordExplode": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): One output row per element of an array. A record whose array is null or empty produces no row.", + "required": ["path", "as"], + "properties": { + "path": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(?:\\.[A-Za-z_][A-Za-z0-9_]*|\\['[^'\\\\\\u0000-\\u001f\\u007f]{1,256}'\\])*(?![\\s\\S])", + "description": "The array: a stream column or a nested field, such as items or order.lines." + }, + "as": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(?![\\s\\S])", + "description": "The element's name in mapping paths: item.sku for a field of a struct element, item for a scalar element." + } + } + }, + "recordMappingColumn": { + "type": "object", + "additionalProperties": false, + "description": "NEW in v0.7.7 (preview): Where one output column, or one nested field of one, is read from.", + "required": ["name", "from"], + "properties": { + "name": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(?:\\.[A-Za-z_][A-Za-z0-9_]*)*(?![\\s\\S])", + "description": "The output column, or a field of a struct output column as a dotted path (customer.id). Mapping a nested field builds that struct from the paths given, so a nested output can be written from fields that are flat or elsewhere in the record." + }, + "from": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*(?:\\.[A-Za-z_][A-Za-z0-9_]*|\\['[^'\\\\\\u0000-\\u001f\\u007f]{1,256}'\\])*(?![\\s\\S])", + "description": "The source: a stream column (value, key or metadata), a nested field (customer.address.country), a map entry (attributes['channel'], null when the key is absent) or the exploded element (item.sku)." + }, + "cast": { + "$ref": "#/$defs/column/properties/type", + "description": "Convert the source to this type: binary to string (UTF-8), or a widening (int to long, float to double, a decimal to a wider precision at the same scale, a timestamp to a finer precision)." + }, + "description": {"type": "string"} + } + }, "dqSpec": { "type": "object", "additionalProperties": false, @@ -3350,6 +3480,14 @@ "type": "object", "additionalProperties": true }, + "stream": { + "$ref": "#/$defs/streamSource", + "description": "NEW in v0.7.7 (preview): The stream the build reads. On a managed Flink build (parameters.flink.management) sql is optional: the lane writes the query from the mapping instead. Elsewhere the stream describes what the build's own sql reads." + }, + "mapping": { + "$ref": "#/$defs/recordMapping", + "description": "NEW in v0.7.7 (preview): How the stream's records become the output's rows. Requires stream, and replaces sql." + }, "tags": { "$ref": "#/$defs/tags" }, @@ -3357,9 +3495,38 @@ "$ref": "#/$defs/labels" } }, - "required": [ - "sql" - ] + "if": { + "not": { + "required": [ + "stream", + "parameters" + ], + "properties": { + "parameters": { + "required": [ + "flink" + ], + "properties": { + "flink": { + "required": [ + "management" + ] + } + } + } + } + } + }, + "then": { + "required": [ + "sql" + ] + }, + "dependentRequired": { + "mapping": [ + "stream" + ] + } }, "multiStagePattern": { "type": "object", diff --git a/fluid_build/streaming/compiler.py b/fluid_build/streaming/compiler.py index 4ea41bc1..4574cb30 100644 --- a/fluid_build/streaming/compiler.py +++ b/fluid_build/streaming/compiler.py @@ -35,6 +35,7 @@ ) from fluid_build.providers._iceberg_catalog_auth import FORBIDDEN_TEXT_PATTERN, parse_catalog_auth from fluid_build.providers._sql_safety import quote_ansi_string_literal, validate_ident +from fluid_build.streaming.mapping import SOURCE_TABLE, bind_by_name, compile_select, parse_stream from fluid_build.streaming.types import LogicalType, columns_sql, parse_fields from fluid_build.util.contract import is_externally_run_build @@ -42,6 +43,11 @@ #: ``parameters.flink`` stays what it was: free-form settings of a job that #: runs elsewhere, which forge-cli does not read. _MANAGED_SINCE = "0.7.7" +#: Products whose own properties.sql the operator lets apply deploy +#: (streaming.policy.refuse_unless_contract_sql_allowed). Defined here, not in +#: streaming.policy, because validate reaches this module and must not import +#: build_runners. +CONTRACT_SQL_ALLOWLIST_ENV = "FLUID_FLINK_CONTRACT_SQL_ALLOWLIST" _NAME = r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$" _TEXT = {"type": "string", "minLength": 1} _POSITIVE = {"type": "integer", "minimum": 1} @@ -54,13 +60,10 @@ "source": { "type": "object", "additionalProperties": False, - "required": ["bootstrapServers", "topic", "groupId", "schema", "format"], + "required": ["bootstrapServers", "groupId"], "properties": { "bootstrapServers": _TEXT, - "topic": _TEXT, "groupId": _TEXT, - "format": {"enum": ["json", "avro-confluent"]}, - "schema": {"type": "array", "minItems": 1, "items": {"type": "object"}}, "schemaRegistryUrl": _TEXT, "startupMode": {"enum": ["earliest-offset", "group-offsets"]}, "eventTime": { @@ -224,6 +227,15 @@ def compile_build(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Flin cfg = managed_config(build) if cfg is None: raise ValueError("This Flink build is externally managed; opt in with parameters.flink") + source = cfg.get("source") + moved = ( + sorted({"topic", "format", "schema"} & set(source)) if isinstance(source, Mapping) else [] + ) + if moved: + raise ValueError( + f"parameters.flink.source.{moved[0]}: the topic, format and schema of the records " + "are declared in properties.stream; parameters.flink.source keeps the connection" + ) errors = sorted( jsonschema.Draft202012Validator(_CONFIG).iter_errors(cfg), key=lambda e: str(e.path) ) @@ -245,10 +257,14 @@ def compile_build(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Flin outputs = build.get("outputs") if not isinstance(outputs, list) or len(outputs) != 1: raise ValueError(f"Flink build {bid}: declare exactly one Iceberg output") - targets = [e for e in contract.get("exposes") or [] if e.get("exposeId") == outputs[0]] + targets = [ + (i, e) + for i, e in enumerate(contract.get("exposes") or []) + if e.get("exposeId") == outputs[0] + ] if len(targets) != 1: raise ValueError(f"Flink build {bid}: output must identify one expose") - target = targets[0] + index, target = targets[0] if ( target.get("policy") or target.get("quality") @@ -290,33 +306,36 @@ def reject_uncompiled_rules(node: Any) -> None: raise ValueError("Iceberg target requires one namespace/database and table") namespace, table = (validate_ident(x) for x in table_parts) fields = parse_fields((target.get("contract") or {}).get("schema")) + props = build.get("properties") or {} + if "stream" not in props: + raise ValueError( + f"Flink build {bid}: managed Flink reads the records properties.stream declares " + "(kind, topic, format, schema)" + ) + spec = props["stream"] + if isinstance(spec, Mapping): + reject_uncompiled_rules(spec.get("schema")) + key = spec.get("key") + if isinstance(key, Mapping): + reject_uncompiled_rules(key.get("schema")) + stream = parse_stream(spec, f"Flink build {bid}: properties.stream") source = cfg["source"] - reject_uncompiled_rules(source["schema"]) - source_fields = parse_fields(source["schema"]) + registry = source.get("schemaRegistryUrl") + if stream.uses_registry: + if not registry: + raise ValueError("An avro-confluent value or key requires source.schemaRegistryUrl") + _url(registry, "source.schemaRegistryUrl") + elif registry: + raise ValueError("schemaRegistryUrl applies only to avro-confluent") source_options = { "connector": "kafka", - "topic": source["topic"], + "topic": stream.topic, "properties.bootstrap.servers": source["bootstrapServers"], "properties.group.id": source["groupId"], "scan.startup.mode": source.get("startupMode", "group-offsets"), "properties.auto.offset.reset": "earliest", - "format": source["format"], + **stream.format_options(registry), } - if source["format"] == "avro-confluent": - if not source.get("schemaRegistryUrl"): - raise ValueError("Avro source requires schemaRegistryUrl") - _url(source["schemaRegistryUrl"], "source.schemaRegistryUrl") - source_options["avro-confluent.url"] = source["schemaRegistryUrl"] - elif source.get("schemaRegistryUrl"): - raise ValueError("schemaRegistryUrl applies only to avro-confluent") - else: - source_options.update( - { - "json.fail-on-missing-field": "false", - "json.ignore-parse-errors": "false", - "json.timestamp-format.standard": "ISO-8601", - } - ) deployment = cfg["deployment"] uris = [deployment[k].rstrip("/") for k in ("checkpointUri", "savepointUri", "haUri")] for uri in uris: @@ -336,26 +355,49 @@ def reject_uncompiled_rules(node: Any) -> None: watermark = "" if event_time: name = validate_ident(event_time["column"]) - dtype = dict(source_fields).get(name) + dtype = stream.columns.get(name) if dtype is None or dtype.kind not in {"timestamp", "timestamp_ltz"}: raise ValueError("eventTime.column must identify a top-level timestamp") + # Flink accepts a rowtime of precision 0 to 3 only, and the stream's default is 6. + if (6 if dtype.precision is None else dtype.precision) > 3: + raise ValueError( + "eventTime.column must be a timestamp of precision 3 or less, such as " + "timestamp_ltz(3): Flink's event time is milliseconds" + ) watermark = ( f",\n WATERMARK FOR `{name}` AS `{name}`" f" - INTERVAL '{event_time['delaySeconds']}' SECOND" ) - sql = (build.get("properties") or {}).get("sql") - if not isinstance(sql, str) or not sql.strip(): - raise ValueError("Flink build requires a SELECT query in properties.sql") - # This gate is intentionally conservative. Flink's matching planner performs - # authoritative semantic validation; Forge must not mistake DDL for a transform. - if not re.match(r"(?is)^\s*SELECT\b", sql) or ";" in sql: - raise ValueError("Managed Flink properties.sql must be a single SELECT without semicolons") - # A /*+ OPTIONS(...) */ hint overrides the source's connector options at query - # time (dynamic table options are on by default), so contract SQL could point - # the Kafka source at another broker or wire a config provider that reads the - # pod's secrets. Flink reads a hint only from the literal "/*+" token. - if "/*+" in sql: - raise ValueError("Managed Flink properties.sql must not contain SQL hints (/*+ ... */)") + sql = props.get("sql") + if sql is not None and props.get("mapping") is not None: + raise ValueError( + f"Flink build {bid}: declare properties.mapping or properties.sql, not both" + ) + if sql is None: + query = compile_select( + stream, + props.get("mapping"), + fields, + where=f"Flink build {bid}: properties.mapping", + target_where=f"exposes[{index}].contract.schema", + ) + else: + # A contract's own SELECT, bound by name to the output columns. + if not isinstance(sql, str) or not sql.strip(): + raise ValueError("Flink build properties.sql must be a SELECT query") + # This gate is intentionally conservative. Flink's matching planner performs + # authoritative semantic validation; Forge must not mistake DDL for a transform. + if not re.match(r"(?is)^\s*SELECT\b", sql) or ";" in sql: + raise ValueError( + "Managed Flink properties.sql must be a single SELECT without semicolons" + ) + # A /*+ OPTIONS(...) */ hint overrides the source's connector options at query + # time (dynamic table options are on by default), so contract SQL could point + # the Kafka source at another broker or wire a config provider that reads the + # pod's secrets. Flink reads a hint only from the literal "/*+" token. + if "/*+" in sql: + raise ValueError("Managed Flink properties.sql must not contain SQL hints (/*+ ... */)") + query = bind_by_name(sql, fields) iceberg = binding.get("icebergConfig") or {} if iceberg.get("writeVersion", 2) != 2: raise ValueError("Managed Flink requires Iceberg format version 2") @@ -427,9 +469,9 @@ def reject_uncompiled_rules(node: Any) -> None: f"CREATE DATABASE IF NOT EXISTS `lake`.`{namespace}`", f"CREATE TABLE IF NOT EXISTS {target_name} (\n {columns_sql(fields)}\n)" f"{partition_sql}\nWITH (\n {_options({'format-version': '2', 'write.format.default': iceberg.get('fileFormat', 'parquet'), **table_properties})}\n)", - f"CREATE TEMPORARY TABLE `source_events` (\n {columns_sql(source_fields)}{watermark}\n)" + f"CREATE TEMPORARY TABLE `{SOURCE_TABLE}` (\n {stream.ddl()}{watermark}\n)" f" WITH (\n {_options(source_options)}\n)", - f"INSERT INTO {target_name}\n{sql.strip()}", + f"INSERT INTO {target_name}\n{query}", ) payload = { "artifactVersion": "flink-iceberg-v1", @@ -537,8 +579,24 @@ def reject_uncompiled_rules(node: Any) -> None: def compile_contract(contract: Mapping[str, Any]) -> tuple[FlinkArtifacts, ...]: + return tuple(artifact for _, artifact in compile_builds(contract)) + + +def compile_builds( + contract: Mapping[str, Any], +) -> tuple[tuple[Mapping[str, Any], FlinkArtifacts], ...]: + """Each managed build with the artifacts compiled from it, paired by position. + + Whatever acts on an artifact checks the build it came from, never one found by + id: two builds may share an id, and an operator's allowlists would then be + judged on the wrong one. + """ builds = managed_builds(contract) artifacts = tuple(compile_build(contract, build) for build in builds) + ids = [a.build_id for a in artifacts] + repeated = sorted({i for i in ids if ids.count(i) > 1}) + if repeated: + raise ValueError(f"Managed Flink build ids must be unique; {repeated[0]!r} is used twice") if builds and ( len(builds) != len(contract.get("builds") or []) or {e.get("exposeId") for e in contract.get("exposes") or []} @@ -569,7 +627,7 @@ def compile_contract(contract: Mapping[str, Any]) -> tuple[FlinkArtifacts, ...]: for b in prefixes[i + 1 :] ): raise ValueError("State storage prefixes must not overlap between managed builds") - return artifacts + return tuple(zip(builds, artifacts, strict=True)) def validation_errors(contract: Mapping[str, Any]) -> list[str]: @@ -583,10 +641,30 @@ def validation_errors(contract: Mapping[str, Any]) -> list[str]: def validation_warnings(contract: Mapping[str, Any]) -> list[str]: """An older contract that asks for the managed lane still validates, as before. - Its build stays run outside forge-cli, so say why nothing will deploy it. + Its build stays run outside forge-cli, so say why nothing will deploy it. On the + preview, a managed build that brings its own SQL is told that apply needs the + operator to allow it. """ if managed_lane_enabled(contract): - return [] + try: + builds = managed_builds(contract) + except (ValueError, KeyError, TypeError): + return [] # validation_errors reports it + warnings = [ + f"Build {build.get('id')}: properties.sql is run only if the operator lists " + f"{contract.get('id')} in {CONTRACT_SQL_ALLOWLIST_ENV}; properties.mapping is " + "checked against the output before deploy and reads nothing but the stream" + for build in builds + if (build.get("properties") or {}).get("sql") is not None + ] + for build in contract.get("builds") or []: + props = build.get("properties") if isinstance(build, Mapping) else None + if isinstance(props, Mapping) and "mapping" in props and build not in builds: + warnings.append( + f"Build {build.get('id')}: properties.mapping is compiled only by the " + "managed Flink lane (parameters.flink.management); nothing reads it here" + ) + return warnings warnings = [] for build in contract.get("builds") or []: props = build.get("properties") if isinstance(build, Mapping) else None diff --git a/fluid_build/streaming/mapping.py b/fluid_build/streaming/mapping.py new file mode 100644 index 00000000..f3f72639 --- /dev/null +++ b/fluid_build/streaming/mapping.py @@ -0,0 +1,571 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""Kafka records to output rows, for the opt-in managed Flink lane. + +``properties.stream`` says what the records are and ``properties.mapping`` how they +become the exposed table's rows. forge-cli writes the job's query from the two: every +identifier passes ``validate_ident`` and every literal ``quote_ansi_string_literal``, so +no contract text reaches the SQL, and the job reads nothing but its own source table. + +Every output column's type and nullability is checked against the path it is read from +before anything deploys. dbt's model contracts run that check against the warehouse; +here it is static, because the contract declares every source type. The source table is +the Flink Kafka connector's own DDL: key fields through ``key.fields`` with +``value.fields-include = 'EXCEPT_KEY'``, metadata as ``METADATA FROM '' VIRTUAL`` +(Flink 1.20 "Available Metadata"), and arrays flattened with ``CROSS JOIN UNNEST``. +Paths into the record follow Estuary's projections and Doris' ``jsonpaths``: one +declared path per output column, never an expression. +""" + +from __future__ import annotations + +import dataclasses +import re +from dataclasses import dataclass +from typing import Any, Mapping, Sequence + +from fluid_build.providers._sql_safety import quote_ansi_string_literal, validate_ident +from fluid_build.streaming.types import LogicalType, columns_sql, parse_fields, parse_type + +SOURCE_TABLE = "source_events" +VALUE_FORMATS = ("json", "avro-confluent") +KEY_FORMATS = ("raw", "json", "avro-confluent") +_SRC = "src" +_ELEM = "elem" +_IDENT = r"[A-Za-z_][A-Za-z0-9_]*" +_ENTRY = r"\['([^'\\\x00-\x1f\x7f]{1,256})'\]" +_PATH = re.compile(rf"{_IDENT}(?:\.{_IDENT}|{_ENTRY})*") +_STEP = re.compile(rf"\.?({_IDENT})|{_ENTRY}") +_TARGET = re.compile(rf"{_IDENT}(?:\.{_IDENT})*") +_TOPIC = re.compile(r"[A-Za-z0-9._-]{1,249}") +MAX_STEPS = 64 + +#: Kafka connector metadata: the DDL type Flink declares, and the logical type it reads as. +_METADATA: dict[str, tuple[str, LogicalType]] = { + "topic": ("STRING NOT NULL", LogicalType("string", True)), + "partition": ("INT NOT NULL", LogicalType("int", True)), + "offset": ("BIGINT NOT NULL", LogicalType("long", True)), + "timestamp": ("TIMESTAMP_LTZ(3) NOT NULL", LogicalType("timestamp_ltz", True, precision=3)), + "timestamp-type": ("STRING NOT NULL", LogicalType("string", True)), + "headers": ( + "MAP NOT NULL", + LogicalType("map", True, key=LogicalType("string", True), value=LogicalType("binary")), + ), + "leader-epoch": ("INT", LogicalType("int")), +} +_JSON_OPTIONS = { + "json.fail-on-missing-field": "false", + "json.ignore-parse-errors": "false", + "json.timestamp-format.standard": "ISO-8601", +} +_WIDENING = {("int", "long"), ("float", "double")} + +Step = tuple[str, str] # ("field", name) or ("entry", map key) + + +@dataclass(frozen=True) +class Stream: + """``properties.stream``, parsed: the columns of the job's source table.""" + + topic: str + format: str + value: tuple[tuple[str, LogicalType], ...] + key_format: str | None = None + key: tuple[tuple[str, LogicalType], ...] = () + key_prefix: str = "" + metadata: tuple[tuple[str, str], ...] = () + + @property + def columns(self) -> dict[str, LogicalType]: + """Every column of the source table, by name: value, key, then metadata.""" + return { + **dict(self.value), + **dict(self.key), + **{name: _METADATA[source][1] for name, source in self.metadata}, + } + + @property + def uses_registry(self) -> bool: + return "avro-confluent" in (self.format, self.key_format) + + def ddl(self) -> str: + parts = [columns_sql(self.value)] + if self.key: + parts.append(columns_sql(self.key)) + parts.extend( + f"`{name}` {_METADATA[source][0]} METADATA FROM " + f"{quote_ansi_string_literal(source)} VIRTUAL" + for name, source in self.metadata + ) + return ",\n ".join(parts) + + def format_options(self, registry_url: str | None) -> dict[str, str]: + """Connector options for the value format and, when declared, the key's.""" + + def options(fmt: str) -> dict[str, str]: + if fmt == "json": + return dict(_JSON_OPTIONS) + if fmt == "avro-confluent" and registry_url: + return {"avro-confluent.url": registry_url} + return {} + + if self.key_format is None: + return {"format": self.format, **options(self.format)} + result = { + "key.format": self.key_format, + "key.fields": ";".join(name for name, _ in self.key), + "value.format": self.format, + "value.fields-include": "EXCEPT_KEY", + **{f"value.{k}": v for k, v in options(self.format).items()}, + **{f"key.{k}": v for k, v in options(self.key_format).items()}, + } + if self.key_prefix: + result["key.fields-prefix"] = self.key_prefix + return result + + +def _object(value: Any, where: str, allowed: set[str]) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{where}: expected an object") + unknown = sorted(str(key) for key in set(value) - allowed) + if unknown: + raise ValueError(f"{where}: unsupported field(s) {unknown}") + return value + + +def parse_stream(spec: Any, where: str = "properties.stream") -> Stream: + stream = _object(spec, where, {"kind", "topic", "format", "schema", "key", "metadata"}) + if stream.get("kind") != "kafka": + raise ValueError(f"{where}.kind: only kafka streams are supported") + topic = stream.get("topic") + if not isinstance(topic, str) or not _TOPIC.fullmatch(topic) or topic in {".", ".."}: + raise ValueError(f"{where}.topic: expected one Kafka topic name") + fmt = stream.get("format") + if fmt not in VALUE_FORMATS: + raise ValueError(f"{where}.format: expected one of {list(VALUE_FORMATS)}") + value = parse_fields(stream.get("schema"), path=f"{where}.schema") + key_format: str | None = None + key: tuple[tuple[str, LogicalType], ...] = () + prefix = "" + if "key" in stream: + spec_key = _object(stream["key"], f"{where}.key", {"format", "schema", "prefix"}) + key_format = spec_key.get("format") + if key_format not in KEY_FORMATS: + raise ValueError(f"{where}.key.format: expected one of {list(KEY_FORMATS)}") + prefix = spec_key.get("prefix", "") + if "prefix" in spec_key and ( + not isinstance(prefix, str) or not re.fullmatch(_IDENT, prefix) + ): + raise ValueError(f"{where}.key.prefix: expected an identifier") + fields = parse_fields(spec_key.get("schema"), path=f"{where}.key.schema") + if key_format == "raw" and ( + len(fields) != 1 or fields[0][1].kind not in {"string", "binary"} + ): + raise ValueError(f"{where}.key: a raw key is one column of type string or binary") + key = tuple((validate_ident(prefix + name), dtype) for name, dtype in fields) + metadata: list[tuple[str, str]] = [] + entries = stream.get("metadata", []) + if not isinstance(entries, list) or ("metadata" in stream and not entries): + raise ValueError(f"{where}.metadata: expected a non-empty list") + for i, entry in enumerate(entries): + item = _object(entry, f"{where}.metadata[{i}]", {"name", "from", "description"}) + name, source = item.get("name"), item.get("from") + if not isinstance(name, str): + raise ValueError(f"{where}.metadata[{i}].name: a column name is required") + if not isinstance(source, str) or source not in _METADATA: + raise ValueError(f"{where}.metadata[{i}].from: expected one of {sorted(_METADATA)}") + metadata.append((validate_ident(name), source)) + seen: set[str] = set() + for name in [n for n, _ in value] + [n for n, _ in key] + [n for n, _ in metadata]: + if name.casefold() in seen: + raise ValueError( + f"{where}: column {name!r} is declared twice across schema, key and metadata" + + ("; set key.prefix" if key else "") + ) + seen.add(name.casefold()) + return Stream(topic, fmt, value, key_format, key, prefix, tuple(metadata)) + + +def parse_path(text: Any, where: str) -> tuple[Step, ...]: + """``customer.address.country`` or ``attributes['channel']``, never SQL.""" + if not isinstance(text, str) or not _PATH.fullmatch(text): + raise ValueError( + f"{where}: expected a path such as customer.address.country or attributes['channel']" + ) + steps = tuple(("field", m[1]) if m[1] else ("entry", m[2]) for m in _STEP.finditer(text)) + if len(steps) > MAX_STEPS: + raise ValueError(f"{where}: a path has at most {MAX_STEPS} steps") + return steps + + +def show(steps: Sequence[Step]) -> str: + text = "" + for kind, name in steps: + text += f"['{name}']" if kind == "entry" else (f".{name}" if text else name) + return text + + +def _nullable(dtype: LogicalType, nullable: bool) -> LogicalType: + return dataclasses.replace(dtype, required=not nullable) + + +def _sql_type(dtype: LogicalType) -> str: + """The type in a CAST: nullability inside it kept, the outer NOT NULL dropped.""" + return _nullable(dtype, True).sql() + + +def _describe(dtype: LogicalType) -> str: + if dtype.kind == "decimal": + return f"decimal({dtype.precision},{dtype.scale})" + if dtype.kind in {"timestamp", "timestamp_ltz"}: + return f"{dtype.kind}({_precision(dtype)})" + return dtype.kind + + +def _precision(dtype: LogicalType) -> int: + return 6 if dtype.precision is None else dtype.precision + + +def _difference(src: LogicalType, tgt: LogicalType, path: str) -> str | None: + """The first way ``src`` cannot be written to ``tgt`` as it is, or ``None``.""" + at = f" at {path}" if path else "" + if tgt.required and not src.required: + return f"a value{at} can be null, but the output requires it" + if src.kind != tgt.kind: + return f"{_describe(src)}{at} is not {_describe(tgt)}" + if src.kind == "struct": + names = [name for name, _ in src.fields] + if names != [name for name, _ in tgt.fields]: + return f"struct fields{at} are {names}, not {[name for name, _ in tgt.fields]}" + for (name, child), (_, other) in zip(src.fields, tgt.fields, strict=True): + found = _difference(child, other, f"{path}.{name}" if path else name) + if found: + return found + return None + if src.kind == "array" and src.element is not None and tgt.element is not None: + return _difference(src.element, tgt.element, f"{path}[]" if path else "[]") + if src.kind == "map" and src.value is not None and tgt.value is not None: + # Iceberg map keys are always required, and both sides' keys are strings. + return _difference(src.value, tgt.value, f"{path}{{value}}" if path else "{value}") + if _describe(src) != _describe(tgt): + return f"{_describe(src)}{at} is not {_describe(tgt)}" + return None + + +def _widens(src: LogicalType, tgt: LogicalType) -> bool: + """A scalar the output can hold without loss: Iceberg's own promotions, and finer time.""" + if (src.kind, tgt.kind) in _WIDENING: + return True + if src.kind == tgt.kind == "decimal": + if src.precision is None or tgt.precision is None: + return False + return src.scale == tgt.scale and tgt.precision > src.precision + if src.kind == tgt.kind and src.kind in {"timestamp", "timestamp_ltz"}: + return _precision(tgt) > _precision(src) + return False + + +@dataclass(frozen=True) +class _Ref: + sql: str + dtype: LogicalType # the outer ``required`` is False when the value can be null + origin: tuple[Step, ...] | None # the path from a source column; None from the element + + +@dataclass(frozen=True) +class _Out: + sql: str + nullable: bool + + +@dataclass(frozen=True) +class _Override: + where: str + source: tuple[Step, ...] + cast: LogicalType | None + + +class _Compiler: + def __init__(self, stream: Stream, spec: Any, where: str, target_where: str) -> None: + self.columns = stream.columns + self.where = where + self.target_where = target_where + self.alias: str | None = None + self.element: LogicalType | None = None + self.alias_used = False + self.join = "" + # Source structs a row being built is known to read with a value: it is null + # where they are, so their own nullability does not reach its fields. + self.present: list[tuple[Step, ...]] = [] + mapping = _object({} if spec is None else spec, where, {"explode", "columns"}) + if "explode" in mapping: + self._explode(_object(mapping["explode"], f"{where}.explode", {"path", "as"})) + self.overrides: dict[str, _Override] = {} + entries = mapping.get("columns", []) + if not isinstance(entries, list) or ("columns" in mapping and not entries): + raise ValueError(f"{where}.columns: expected a non-empty list") + for i, entry in enumerate(entries): + at = f"{where}.columns[{i}]" + item = _object(entry, at, {"name", "from", "cast", "description"}) + name = item.get("name") + if not isinstance(name, str) or not _TARGET.fullmatch(name): + raise ValueError(f"{at}.name: expected an output column or a dotted path into one") + if name in self.overrides: + raise ValueError(f"{at}.name: {name} is mapped twice") + cast = None + if "cast" in item: + if str(item["cast"]).strip().lower() in {"struct", "row", "record", "array", "map"}: + raise ValueError(f"{at}.cast: only a scalar type can be a conversion target") + cast = parse_type({"type": item["cast"]}, path=f"{at}.cast") + self.overrides[name] = _Override(at, parse_path(item.get("from"), f"{at}.from"), cast) + # Each struct output path with a mapped field under it, and the first such field, + # found once instead of scanning every override for every output path. + self.inside: dict[str, str] = {} + for name in self.overrides: + parts = name.split(".") + for end in range(1, len(parts)): + self.inside.setdefault(".".join(parts[:end]), name) + self.used: set[str] = set() + + def _explode(self, spec: Mapping[str, Any]) -> None: + at = f"{self.where}.explode" + array = self.resolve(parse_path(spec.get("path"), f"{at}.path"), f"{at}.path") + if array.dtype.kind != "array" or array.dtype.element is None: + raise ValueError(f"{at}.path: {show(array.origin or ())} is not an array") + alias = spec.get("as") + if not isinstance(alias, str) or not re.fullmatch(_IDENT, alias): + raise ValueError(f"{at}.as: expected an identifier") + if alias in self.columns: + raise ValueError(f"{at}.as: {alias} is already a column of the stream") + element = array.dtype.element + names = [name for name, _ in element.fields] if element.kind == "struct" else [alias] + self.alias, self.element = alias, element + self.join = ( + f"\nCROSS JOIN UNNEST({array.sql}) AS `{_ELEM}` " + f"({', '.join(f'`{name}`' for name in names)})" + ) + + def resolve(self, steps: tuple[Step, ...], where: str) -> _Ref: + root = steps[0][1] + rest = steps[1:] + done = 1 + if self.alias is not None and self.element is not None and root == self.alias: + self.alias_used = True + element, nullable = self.element, not self.element.required + if element.kind != "struct": + ref = _Ref(f"`{_ELEM}`.`{root}`", _nullable(element, nullable), None) + elif rest and rest[0][0] == "field": + by_name = dict(element.fields) + if rest[0][1] not in by_name: + raise ValueError(f"{where}: the element of {root} has no field {rest[0][1]}") + child = by_name[rest[0][1]] + ref = _Ref( + f"`{_ELEM}`.`{rest[0][1]}`", + _nullable(child, nullable or not child.required), + None, + ) + rest = rest[1:] + done = 2 + else: + # UNNEST spreads a struct element over columns, so the whole element is + # rebuilt from them. A rebuilt row is never null; its fields carry the nulls. + fields = tuple( + (name, _nullable(child, nullable or not child.required)) + for name, child in element.fields + ) + rebuilt = LogicalType("struct", True, fields=fields) + cells = ", ".join(f"`{_ELEM}`.`{name}`" for name, _ in fields) + ref = _Ref(f"CAST(ROW({cells}) AS {_sql_type(rebuilt)})", rebuilt, None) + elif root in self.columns: + ref = self._settle(_Ref(f"`{_SRC}`.`{root}`", self.columns[root], (steps[0],))) + else: + raise ValueError(f"{where}: {root} is not a column of properties.stream") + for step in rest: + ref = self._settle(self._step(ref, step, where, show(steps[:done]))) + done += 1 + return ref + + def _settle(self, ref: _Ref) -> _Ref: + if ref.origin is not None and ref.origin in self.present: + return dataclasses.replace(ref, dtype=_nullable(ref.dtype, False)) + return ref + + def _step(self, ref: _Ref, step: Step, where: str, label: str) -> _Ref: + kind, name = step + dtype = ref.dtype + origin = None if ref.origin is None else (*ref.origin, step) + if kind == "field": + if dtype.kind != "struct": + raise ValueError(f"{where}: {label} is {_describe(dtype)}, not a struct") + fields = dict(dtype.fields) + if name not in fields: + raise ValueError(f"{where}: {label} has no field {name}") + child = fields[name] + nullable = not dtype.required or not child.required + return _Ref(f"{ref.sql}.`{name}`", _nullable(child, nullable), origin) + if dtype.kind != "map" or dtype.value is None: + raise ValueError(f"{where}: {label} is {_describe(dtype)}, not a map") + # A key the record does not carry reads as null. + literal = quote_ansi_string_literal(name) + return _Ref(f"{ref.sql}[{literal}]", _nullable(dtype.value, True), origin) + + def _assign(self, ref: _Ref, target: LogicalType, override: _Override | None, name: str) -> str: + where = override.where if override else f"{self.target_where}.{name}" + source = ref.dtype + sql = ref.sql + if override is not None and override.cast is not None: + cast = override.cast + convertible = ( + (source.kind, cast.kind) == ("binary", "string") + or _widens(source, cast) + or _difference(_nullable(source, True), _nullable(cast, True), "") is None + ) + if not convertible: + raise ValueError( + f"{where}.cast: {_describe(source)} cannot become {_describe(cast)}; " + "supported: binary to string, and widening" + ) + sql = f"CAST({sql} AS {_sql_type(cast)})" + source = _nullable(cast, not source.required) + difference = _difference(source, target, "") + if difference is None: + return sql + if _widens(source, target) and (source.required or not target.required): + return f"CAST({sql} AS {_sql_type(target)})" + source_path = show(override.source) if override else f"the stream's {name}" + raise ValueError(f"{where}: {name} cannot be read from {source_path}: {difference}") + + def target(self, path: tuple[str, ...], dtype: LogicalType) -> _Out: + name = ".".join(path) + override = self.overrides.get(name) + nested = self.inside.get(name) + if override is not None: + if nested: + raise ValueError(f"{override.where}: {nested} is inside {name}, mapped whole") + self.used.add(name) + ref = self.resolve(override.source, f"{override.where}.from") + sql = self._assign(ref, dtype, override, name) + return _Out(sql, not ref.dtype.required) + if dtype.kind == "struct" and nested: + return self._row(path, dtype) + steps = tuple(("field", part) for part in path) + try: + ref = self.resolve(steps, f"{self.target_where}.{name}") + except ValueError as exc: + raise ValueError(f"{exc}; map {name} in properties.mapping.columns") from None + if ( + dtype.kind == "struct" + and ref.dtype.kind == "struct" + and _difference(ref.dtype, dtype, "") is not None + ): + return self._row(path, dtype) + return _Out(self._assign(ref, dtype, None, name), not ref.dtype.required) + + def _row(self, path: tuple[str, ...], dtype: LogicalType) -> _Out: + """A struct output built field by field, from the paths its fields are read from.""" + planned = tuple( + origin for name, child in dtype.fields for origin in self._plan((*path, name), child) + ) + parent = self._shared_struct(planned) + if parent is not None and dtype.required: + raise ValueError( + f"{self.target_where}.{'.'.join(path)}: required, but its fields are read " + f"from {show(parent.origin or ())}, which can be null" + ) + if parent is not None and parent.origin is not None: + self.present.append(parent.origin) + try: + children = [self.target((*path, name), child) for name, child in dtype.fields] + finally: + if parent is not None and parent.origin is not None: + self.present.pop() + row = f"ROW({', '.join(child.sql for child in children)})" + if parent is None: + return _Out(f"CAST({row} AS {_sql_type(dtype)})", False) + # Fields read from one nullable struct: the output is null where that struct is. + # Flink 1.20's validator refuses a CAST(ROW(...)) as a CASE branch, so the bare + # ROW is the branch and the CASE is cast to the output's type. + case = f"CASE WHEN {parent.sql} IS NULL THEN NULL ELSE {row} END" + return _Out(f"CAST({case} AS {_sql_type(dtype)})", True) + + def _plan( + self, path: tuple[str, ...], dtype: LogicalType + ) -> tuple[tuple[Step, ...] | None, ...]: + """Where each field of a row to be built is read from, before any of it is checked.""" + name = ".".join(path) + override = self.overrides.get(name) + if override is not None: + steps = override.source + elif dtype.kind == "struct" and name in self.inside: + return tuple(o for child, t in dtype.fields for o in self._plan((*path, child), t)) + else: + steps = tuple(("field", part) for part in path) + return (steps if steps[0][1] in self.columns and steps[0][1] != self.alias else None,) + + def _shared_struct(self, origins: tuple[tuple[Step, ...] | None, ...]) -> _Ref | None: + """The nullable source struct every field of a built row is read from, if one is.""" + paths = [origin for origin in origins if origin is not None] + if not paths or len(paths) != len(origins): + return None + prefix = list(paths[0]) + for path in paths[1:]: + common = 0 + while common < min(len(prefix), len(path)) and prefix[common] == path[common]: + common += 1 + prefix = prefix[:common] + while prefix: + if all(len(path) > len(prefix) for path in paths): + try: + ref = self.resolve(tuple(prefix), self.where) + except ValueError: + return None # the fields' own paths report it + if ref.dtype.kind == "struct": + return None if ref.dtype.required else ref + prefix.pop() + return None + + +def compile_select( + stream: Stream, + mapping: Any, + target: Sequence[tuple[str, LogicalType]], + *, + where: str = "properties.mapping", + target_where: str = "exposes.contract.schema", +) -> str: + """The query forge-cli writes: every output column in the expose's order, checked.""" + compiler = _Compiler(stream, mapping, where, target_where) + select = [f"{compiler.target((name, ), dtype).sql} AS `{name}`" for name, dtype in target] + unused = sorted(set(compiler.overrides) - compiler.used) + if unused: + raise ValueError( + f"{compiler.overrides[unused[0]].where}.name: {unused[0]} is not an output column " + "or a struct field of one (a field inside an array or map is mapped with its column)" + ) + if compiler.alias is not None and not compiler.alias_used: + raise ValueError(f"{where}.explode.as: no output column reads {compiler.alias}") + return ( + "SELECT\n " + ",\n ".join(select) + f"\nFROM `{SOURCE_TABLE}` AS `{_SRC}`" + compiler.join + ) + + +def bind_by_name(sql: str, target: Sequence[tuple[str, LogicalType]]) -> str: + """A contract's own SELECT, read by output column name instead of by position. + + A column it does not produce then fails Flink's planner instead of shifting the + others; the planner, in the pod, also checks the types. + """ + columns = ", ".join(f"`{name}`" for name, _ in target) + return f"SELECT {columns}\nFROM (\n{sql.strip()}\n) AS `contract_sql`" diff --git a/fluid_build/streaming/policy.py b/fluid_build/streaming/policy.py index 49bf27dd..429fc5bd 100644 --- a/fluid_build/streaming/policy.py +++ b/fluid_build/streaming/policy.py @@ -52,6 +52,7 @@ from __future__ import annotations +import hashlib import ipaddress import os from dataclasses import dataclass @@ -73,6 +74,7 @@ from fluid_build.build_runners._secret_policy import allowlist as secret_ref_allowlist from fluid_build.build_runners._secret_policy import is_loopback, variable_allowed from fluid_build.providers._iceberg_catalog_auth import CatalogAuth, parse_catalog_auth +from fluid_build.streaming.compiler import CONTRACT_SQL_ALLOWLIST_ENV CONTEXT_ALLOWLIST_ENV = "FLUID_FLINK_KUBE_CONTEXT_ALLOWLIST" NAMESPACE_ALLOWLIST_ENV = "FLUID_FLINK_NAMESPACE_ALLOWLIST" @@ -338,3 +340,60 @@ def refuse_unless_catalog_targets_allowed( refuse_unless_catalog_secret_allowed(auth.env_var, str(token_endpoint), environ=environ) refuse_unless_catalog_secret_allowed(auth.env_var, catalog_uri, environ=environ) return auth + + +def contract_sql_entry(product_id: Any, build: Mapping[str, Any]) -> str: + """The allowlist entry that approves exactly this build's SQL: ``product/build@sha256:..``.""" + sql = (build.get("properties") or {}).get("sql") + digest = hashlib.sha256(str(sql).strip().encode("utf-8")).hexdigest() + return f"{product_id}/{build.get('id')}@sha256:{digest}" + + +def refuse_unless_contract_sql_allowed( + product_id: Any, build: Mapping[str, Any], environ: Optional[Mapping[str, str]] = None +) -> None: + """Raise unless the operator lists a build whose own SQL the job would run. + + forge-cli writes the query from ``properties.mapping``, and that query reads only + the stream. A contract's ``properties.sql`` runs inside the job with its catalog + identity, so it can read any table that identity can (``lake..``) + and write it into the product's own table. + + An entry is one of: + - ``/@sha256:``, which approves + the reviewed SQL only, so an edit needs a new review (recommended); + - ``/``, which approves the build's SQL as it changes; + - ````, which approves every build of the product. + """ + props = build.get("properties") + if not isinstance(props, Mapping) or props.get("sql") is None: + return + build_id = build.get("id") + pinned = contract_sql_entry(product_id, build) + entries = _entries(CONTRACT_SQL_ALLOWLIST_ENV, environ) + if {str(product_id), f"{product_id}/{build_id}", pinned} & set(entries): + return + stale = any(e.startswith(f"{product_id}/{build_id}@") for e in entries) + listed = ",".join( + [*(e for e in entries if not e.startswith(f"{product_id}/{build_id}@")), pinned] + ) + raise FlinkPolicyError( + what=f"managed Flink build {build_id!r} runs SQL written in the contract", + why=( + ( + f"{CONTRACT_SQL_ALLOWLIST_ENV} approves other SQL for {product_id}/{build_id}: " + "the build's SQL changed after it was listed. " + if stale + else f"{product_id}/{build_id} is not in {CONTRACT_SQL_ALLOWLIST_ENV}. " + ) + + "A contract's own SELECT runs inside the job with its catalog identity: it can " + "read any table that identity can read, not only the stream, and copy it into " + "this product's table. An unset list allows nothing." + ), + fix=( + "Declare the output in properties.mapping, which forge-cli compiles into a query " + "that reads only the stream; or, once this SQL is reviewed, list it " + f'{_OPERATOR_ONLY}, pinned to what was reviewed: export {CONTRACT_SQL_ALLOWLIST_ENV}="{listed}".' + ), + doc=doc_url(), + ) diff --git a/fluid_build/streaming/runtime.py b/fluid_build/streaming/runtime.py index e161869c..aa78dac8 100644 --- a/fluid_build/streaming/runtime.py +++ b/fluid_build/streaming/runtime.py @@ -29,10 +29,11 @@ import httpx from fluid_build.providers._iceberg_catalog import resolve_iceberg_catalog -from fluid_build.streaming.compiler import FlinkArtifacts, compile_contract, managed_config +from fluid_build.streaming.compiler import FlinkArtifacts, compile_builds, managed_config from fluid_build.streaming.policy import ( catalog_request_guard, refuse_unless_catalog_targets_allowed, + refuse_unless_contract_sql_allowed, refuse_unless_deployment_allowed, ) from fluid_build.streaming.types import LogicalType @@ -178,7 +179,7 @@ def reconcile( dry_run: bool = False, timeout_seconds: float = 300, ) -> list[dict[str, Any]]: - artifacts = compile_contract(contract) + compiled = compile_builds(contract) if mode.startswith("replace"): raise StreamingError( "replace/reset is not supported for managed streams; restore state explicitly" @@ -188,10 +189,9 @@ def reconcile( dry_run = dry_run or mode == "dry-run" if not math.isfinite(timeout_seconds) or timeout_seconds <= 0: raise StreamingError("Readiness timeout must be positive") - builds = {b["id"]: b for b in contract.get("builds") or []} pending: list[tuple[FlinkArtifacts, Kubectl, dict[str, Any] | None]] = [] - for artifact in artifacts: - build = builds[artifact.build_id] + for build, artifact in compiled: + refuse_unless_contract_sql_allowed(contract.get("id"), build) client = _client(build) desired = artifact.resources[1] name = desired["metadata"]["name"] @@ -453,10 +453,9 @@ def catalog_evidence( def inspect_drift(contract: Mapping[str, Any]) -> list[dict[str, Any]]: """Compare live resources to their immutable last-applied spec, not the next plan.""" - builds = {b["id"]: b for b in contract.get("builds") or []} results = [] - for artifact in compile_contract(contract): - client = _client(builds[artifact.build_id]) + for build, artifact in compile_builds(contract): + client = _client(build) name = artifact.resources[1]["metadata"]["name"] current = client.get("flinkdeployment", name) if current is None: @@ -500,9 +499,7 @@ def inspect_drift(contract: Mapping[str, Any]) -> list[dict[str, Any]]: def _verify_once(contract: Mapping[str, Any]) -> list[dict[str, Any]]: results = [] - builds = {b["id"]: b for b in contract.get("builds") or []} - for artifact in compile_contract(contract): - build = builds[artifact.build_id] + for build, artifact in compile_builds(contract): cfg = managed_config(build) if cfg is None: raise StreamingError("Expected managed configuration") diff --git a/tests/integration/test_flink_iceberg_live.py b/tests/integration/test_flink_iceberg_live.py index 656806ea..052a04d6 100644 --- a/tests/integration/test_flink_iceberg_live.py +++ b/tests/integration/test_flink_iceberg_live.py @@ -22,6 +22,7 @@ import copy import json +import logging import os import shutil import subprocess @@ -34,6 +35,9 @@ import yaml from fluid_build.streaming.compiler import compile_contract +from fluid_build.streaming.types import columns_sql + +_LOG = logging.getLogger(__name__) pytestmark = [ pytest.mark.integration, @@ -74,7 +78,7 @@ def test_generated_flink_job_preserves_nested_values_after_task_restart(tmp_path cfg = contract["builds"][0]["properties"]["parameters"]["flink"] cfg["source"]["startupMode"] = "earliest-offset" cfg["catalogProperties"] = { - "s3.endpoint": "http://minio:9000", + "s3.endpoint": "http://s3:9090", "s3.path-style-access": "true", "client.region": "us-east-1", } @@ -232,7 +236,7 @@ def compose(*args, input=None, timeout=180): from datetime import datetime from pyiceberg.catalog.rest import RestCatalog cat = RestCatalog("test", uri="http://iceberg-rest:8181", - **{"s3.endpoint":"http://minio:9000", "s3.access-key-id":"minioadmin", + **{"s3.endpoint":"http://s3:9090", "s3.access-key-id":"minioadmin", "s3.secret-access-key":"minioadmin", "s3.region":"us-east-1"}) rows = [] for _ in range(90): @@ -281,7 +285,11 @@ def compose(*args, input=None, timeout=180): # Existing tables must be checked before INSERT planning/submission, # including changes the planner could otherwise coerce implicitly. drifted = copy.deepcopy(contract) + # The contract agrees with itself (stream and output both int), so it compiles; + # only the table the first run created disagrees, and the runner must refuse it. drifted["exposes"][0]["contract"]["schema"][1]["fields"][1]["fields"][0]["type"] = "int" + stream = drifted["builds"][0]["properties"]["stream"] + stream["schema"][1]["fields"][1]["fields"][0]["type"] = "int" (mismatch,) = compile_contract(drifted) (tmp_path / "job.properties").write_text(mismatch.resources[0]["data"]["job.properties"]) rejected = subprocess.run( @@ -313,3 +321,395 @@ def compose(*args, input=None, timeout=180): ) finally: compose("down", "-v", "--remove-orphans") + + +#: The shapes example: two products reading one stream fragment. +_SHAPES = ("order-items", "orders-nested") + + +def _shape_events(): + """Order events that exercise every null the mappings have to carry.""" + events = [] + for i in range(60): + customer = ( + None + if i % 7 == 0 + else { + "id": f"c-{i}", + "address": ( + None + if i % 5 == 0 + else { + "country": "DE", + "geo": { + "location": {"coordinates": {"latitude": 52.52, "longitude": 13.405}} + }, + } + ), + } + ) + items = ( + None + if i % 13 == 0 + else ( + [{"sku": f"p-{i}", "price": "19.95"}, {"sku": None, "price": None}, None] + if i % 2 + else [] + ) + ) + attributes = ( + None if i % 11 == 0 else ({"channel": "mobile"} if i % 3 else {"campaign": "x"}) + ) + events.append( + { + "event_id": f"evt-{i}", + "customer": customer, + "items": items, + "attributes": attributes, + "event_time": "2026-01-01T00:00:01.123Z", + } + ) + return events + + +def test_generated_flink_jobs_flatten_and_reshape_keyed_events(tmp_path): + """Both shapes, from one stream fragment: rows per item, and a nested reshape. + + The records carry a key and a header, so the key column, Kafka metadata, + header lookup, struct lifts, map lookup, explode and struct rebuilding are all + read back from Iceberg and compared with what was produced. + """ + if shutil.which("docker") is None: + pytest.skip("Docker is required") + image = os.environ.get("FLUID_FLINK_TEST_IMAGE", "forge-flink:test") + if subprocess.run(["docker", "image", "inspect", image], capture_output=True).returncode: + pytest.skip("Build the example Flink runtime image first") + from fluid_build._contract_loader import load_contract_with_overlay + + root = Path(__file__).resolve().parents[2] + example = root / "examples/flink-iceberg" + jobs = {} + for shape in _SHAPES: + contract = load_contract_with_overlay( + str(example / "shapes" / shape / "contract.fluid.yaml"), None, _LOG + ) + cfg = contract["builds"][0]["properties"]["parameters"]["flink"] + cfg["source"]["startupMode"] = "earliest-offset" + cfg["catalogProperties"] = { + "s3.endpoint": "http://s3:9090", + "s3.path-style-access": "true", + "client.region": "us-east-1", + } + (artifact,) = compile_contract(contract) + jobs[shape] = artifact.resources[0]["data"]["job.properties"] + shutil.copyfile(example / "compose.yaml", tmp_path / "compose.yaml") + project = "forge-flink-shapes-" + uuid.uuid4().hex[:12] + base = ["docker", "compose", "-p", project, "-f", str(tmp_path / "compose.yaml")] + + def compose(*args, input=None, timeout=180, check=True): + result = subprocess.run( + [*base, *args], input=input, text=True, capture_output=True, timeout=timeout + ) + if check: + assert result.returncode == 0, result.stdout + result.stderr + return result + + (tmp_path / "job.properties").write_text(jobs[_SHAPES[0]]) + try: + compose("up", "-d", timeout=300) + port = compose("port", "jobmanager", "8081").stdout.strip().rsplit(":", 1)[1] + rest_port = compose("port", "iceberg-rest", "8181").stdout.strip().rsplit(":", 1)[1] + flink = f"http://127.0.0.1:{port}" + catalog = f"http://127.0.0.1:{rest_port}" + _wait_json(flink + "/overview", lambda x: x.get("slots-available", 0) >= 2) + _wait_json(catalog + "/v1/config", lambda x: "defaults" in x) + deadline = time.monotonic() + 60 + while compose( + "exec", "-T", "kafka", "rpk", "topic", "create", "orders", check=False + ).returncode: + assert time.monotonic() < deadline, "the orders topic was never created" + time.sleep(1) + for shape in _SHAPES: + # The job reads job.properties when it is submitted, so the next one can follow. + (tmp_path / "job.properties").write_text(jobs[shape]) + compose( + "exec", + "-T", + "jobmanager", + "flink", + "run", + "-d", + "-c", + "io.fluid.forge.ForgeSqlRunner", + "/opt/flink/usrlib/forge-sql-runner.jar", + "/etc/forge/job.properties", + ) + _wait_json( + flink + "/jobs/overview", + lambda x: sum(j["state"] == "RUNNING" for j in x.get("jobs", [])) == 2, + ) + events = _shape_events() + # One key per record (rpk's %k %v: key, a space, then the value), one header + # per batch. Compact JSON keeps the value free of spaces. + for batch, header in ((events[:30], "trace-a"), (events[30:], "trace-b")): + compose( + "exec", + "-T", + "kafka", + "rpk", + "topic", + "produce", + "orders", + "-f", + "%k %v\n", + "-H", + f"trace-id:{header}", + input="".join( + f"key-{e['event_id']} {json.dumps(e, separators=(',', ':'))}\n" for e in batch + ), + ) + for table in ("order_items", "orders_nested"): + _wait_json( + catalog + f"/v1/namespaces/commerce/tables/{table}", + lambda x: x.get("metadata", {}).get("current-snapshot-id", -1) != -1, + ) + readback = """ +import json +import time +from decimal import Decimal +from datetime import datetime +from pyiceberg.catalog.rest import RestCatalog +cat = RestCatalog("test", uri="http://iceberg-rest:8181", + **{"s3.endpoint":"http://s3:9090", "s3.access-key-id":"minioadmin", + "s3.secret-access-key":"minioadmin", "s3.region":"us-east-1"}) +events = json.loads(EVENTS) +when = datetime.fromisoformat("2026-01-01T00:00:01.123+00:00") +offset = {e["event_id"]: i for i, e in enumerate(events)} +trace = {e["event_id"]: ("trace-a" if i < 30 else "trace-b") for i, e in enumerate(events)} + +def rows(table, count): + found = [] + for _ in range(90): + found = cat.load_table(f"commerce.{table}").scan().to_arrow().to_pylist() + if len(found) >= count: + break + time.sleep(1) + return found + +def address(e): + return (e["customer"] or {}).get("address") or {} + +def coordinates(e): + return ((address(e).get("geo") or {}).get("location") or {}).get("coordinates") + +expected_items = [ + { + "event_id": e["event_id"], + "order_key": "key-" + e["event_id"], + "customer_id": (e["customer"] or {}).get("id"), + "country": address(e).get("country"), + "latitude": (coordinates(e) or {}).get("latitude"), + "longitude": (coordinates(e) or {}).get("longitude"), + "channel": (e["attributes"] or {}).get("channel"), + # A null element reads as a row whose element fields are null. + "sku": (item or {}).get("sku"), + "price": None if (item or {}).get("price") is None else Decimal(item["price"]), + "trace_id": trace[e["event_id"]], + "kafka_partition": 0, + "kafka_offset": offset[e["event_id"]], + "event_time": when, + } + for e in events + for item in e["items"] or [] +] +items = rows("order_items", len(expected_items)) +key = lambda r: (r["event_id"], r["sku"] or "") +assert sorted(items, key=key) == sorted(expected_items, key=key), (len(items), len(expected_items)) + +nested = rows("orders_nested", len(events)) +assert len(nested) == len(events), len(nested) +for row in nested: + e = events[offset[row["event_id"]]] + stamp = row["kafka"].pop("timestamp") + assert stamp is not None + assert row["order"] == { + "key": "key-" + e["event_id"], + "channel": (e["attributes"] or {}).get("channel"), + "placed_at": when, + }, row + assert row["customer"] == ( + None if e["customer"] is None + else {"id": e["customer"]["id"], "country": address(e).get("country")} + ), row + assert row["location"] == coordinates(e), row + expected = e["items"] + if expected is not None: + expected = [ + None if i is None + else {"sku": i["sku"], "price": None if i["price"] is None else Decimal(i["price"])} + for i in expected + ] + assert row["items"] == expected, row + assert row["kafka"] == {"partition": 0, "offset": offset[e["event_id"]]}, row + assert row["event_time"] == when, row +print(f"PASS: {len(items)} item rows, {len(nested)} nested rows; key, header, metadata, nulls") +""" + readback = readback.replace("EVENTS", repr(json.dumps(events))) + result = subprocess.run( + [ + "docker", + "run", + "--rm", + "--network", + project + "_default", + "python:3.11-slim", + "sh", + "-ec", + "pip -q install 'pyiceberg[s3fs,pyarrow]==0.10.0' && python -", + ], + input=readback, + text=True, + capture_output=True, + timeout=240, + ) + assert result.returncode == 0, result.stdout + result.stderr + finally: + compose("down", "-v", "--remove-orphans", check=False) + + +def _column(name, type_, **kw): + return {"name": name, "type": type_, **kw} + + +def _planner_cases(): + """Every example, plus the mappings whose SQL is hardest on Flink's validator.""" + from fluid_build._contract_loader import load_contract_with_overlay + + example = Path(__file__).resolve().parents[2] / "examples/flink-iceberg" + base = yaml.safe_load((example / "contract.fluid.yaml").read_text()) + cases = {"example": base} + for shape in _SHAPES: + cases[shape] = load_contract_with_overlay( + str(example / "shapes" / shape / "contract.fluid.yaml"), None, _LOG + ) + + def variant(name, columns, mapping=None, stream=None): + contract = copy.deepcopy(base) + props = contract["builds"][0]["properties"] + if stream is not None: + stream(props["stream"]) + contract["exposes"][0]["contract"]["schema"] = columns + if mapping is not None: + props["mapping"] = mapping + cases[name] = contract + + def required_fields(stream): + customer = stream["schema"][1] + customer["fields"][0]["required"] = True + customer["fields"].append(_column("visits", "int", required=True)) + + event_id = _column("event_id", "string", required=True) + variant( + "required fields of an optional struct, widened", + [ + event_id, + _column( + "customer", + "struct", + fields=[ + _column("visits", "long", required=True), + _column("id", "string", required=True), + ], + ), + ], + stream=required_fields, + ) + variant( + "a null-propagated struct inside another", + [ + event_id, + _column( + "customer", + "struct", + fields=[ + _column("id", "string"), + _column("address", "struct", fields=[_column("country", "string")]), + ], + ), + ], + ) + variant( + "the headers map whole", + [ + event_id, + _column( + "hdrs", + "map", + key={"type": "string", "required": True}, + value={"type": "binary"}, + ), + ], + stream=lambda s: s.update(metadata=[{"name": "hdrs", "from": "headers"}]), + ) + variant( + "an exploded element kept nested", + [ + event_id, + _column( + "item", + "struct", + fields=[_column("sku", "string"), _column("price", "decimal(18,2)")], + ), + ], + mapping={"explode": {"path": "items", "as": "item"}}, + ) + sql = copy.deepcopy(base) + sql["builds"][0]["properties"]["sql"] = "SELECT * FROM source_events" + cases["contract SQL, bound by name"] = sql + return cases + + +def test_generated_queries_plan_in_flink(tmp_path): + """The runtime image's own planner accepts every query forge-cli writes. + + The job's statements need a Kafka broker and an Iceberg catalog to run, but not + to plan: the source table is the generated Kafka DDL, and the output is a + blackhole table with the expose's columns, so EXPLAIN validates the generated + query exactly as the job would submit it. + """ + if shutil.which("docker") is None: + pytest.skip("Docker is required") + image = os.environ.get("FLUID_FLINK_TEST_IMAGE", "forge-flink:test") + if subprocess.run(["docker", "image", "inspect", image], capture_output=True).returncode: + pytest.skip("Build the example Flink runtime image first") + refused = [] + for name, contract in _planner_cases().items(): + (artifact,) = compile_contract(contract) + source, insert = artifact.statements[3], artifact.statements[4] + sink = ( + f"CREATE TEMPORARY TABLE `sink` (\n {columns_sql(artifact.target_fields)}\n)" + " WITH ('connector' = 'blackhole')" + ) + query = "EXPLAIN INSERT INTO `sink`\n" + insert.split("\n", 1)[1] + (tmp_path / "plan.sql").write_text(";\n".join([source, sink, query]) + ";\n") + run = subprocess.run( + [ + "docker", + "run", + "--rm", + "-v", + f"{tmp_path / 'plan.sql'}:/plan.sql:ro", + image, + "/opt/flink/bin/sql-client.sh", + "-f", + "/plan.sql", + ], + capture_output=True, + text=True, + timeout=180, + ) + out = run.stdout + run.stderr + if run.returncode or "[ERROR]" in out or "== Optimized Physical Plan ==" not in out: + reason = next((line for line in out.splitlines() if "Exception" in line), out[-400:]) + refused.append(f"{name}: {reason.strip()[:400]}") + assert not refused, "\n".join(refused) diff --git a/tests/streaming/test_compiler.py b/tests/streaming/test_compiler.py index cf234e0c..30046401 100644 --- a/tests/streaming/test_compiler.py +++ b/tests/streaming/test_compiler.py @@ -20,6 +20,10 @@ from fluid_build.streaming.compiler import compile_contract, managed_config, validation_errors +def _stream(contract): + return contract["builds"][0]["properties"]["stream"] + + def test_deterministic_artifacts_and_native_table_creation(contract): (first,) = compile_contract(contract) (second,) = compile_contract(copy.deepcopy(contract)) @@ -44,7 +48,10 @@ def test_deterministic_artifacts_and_native_table_creation(contract): def test_sql_and_runtime_changes_change_digest(contract): (original,) = compile_contract(contract) - contract["builds"][0]["properties"]["sql"] += " WHERE event_id <> ''" + contract["builds"][0]["properties"]["mapping"] = { + "columns": [{"name": "event_time", "from": "event_time", "cast": "timestamp_ltz(6)"}] + } + contract["exposes"][0]["contract"]["schema"][-1]["type"] = "timestamp_ltz(6)" (changed,) = compile_contract(contract) assert original.digest != changed.digest cfg = managed_config(contract["builds"][0]) @@ -59,8 +66,9 @@ def test_sql_and_runtime_changes_change_digest(contract): (lambda c, f: f["deployment"].update(image="flink:latest"), "invalid configuration"), (lambda c, f: f["deployment"].pop("context"), "invalid configuration"), (lambda c, f: f.update(typo=True), "invalid configuration"), - (lambda c, f: f["source"].update(format="protobuf"), "invalid configuration"), - (lambda c, f: f["source"].update(format="avro-confluent"), "schemaRegistryUrl"), + (lambda c, f: _stream(c).update(format="protobuf"), "format: expected one of"), + (lambda c, f: _stream(c).update(format="avro-confluent"), "schemaRegistryUrl"), + (lambda c, f: f["source"].update(topic="orders"), "declared in properties.stream"), ( lambda c, f: f["deployment"].update(savepointUri=f["deployment"]["checkpointUri"]), "overlap", @@ -117,9 +125,6 @@ def test_catalog_auth_is_reference_only(contract): def test_event_time_and_partition(contract): cfg = managed_config(contract["builds"][0]) cfg["source"]["eventTime"] = {"column": "event_time", "delaySeconds": 5} - contract["builds"][0]["properties"][ - "sql" - ] = "SELECT event_id, customer, items, attributes, event_time FROM source_events" contract["exposes"][0]["binding"]["icebergConfig"]["partitionSpec"] = [ {"sourceColumn": "event_time", "transform": "day"}, ] @@ -136,7 +141,7 @@ def test_uncompiled_governance_is_rejected(contract, surface): elif surface == "column": target["schema"][1]["fields"][0]["validationRules"] = {"unique": True} elif surface == "source": - managed_config(contract["builds"][0])["source"]["schema"][0]["quality"] = ["check"] + _stream(contract)["schema"][0]["quality"] = ["check"] else: target["dq"] = {"tests": ["check"]} with pytest.raises(ValueError, match="enforce|policy/quality"): @@ -202,7 +207,8 @@ def test_null_management_config_is_not_silently_external(contract): (lambda c, f: c["exposes"][0]["binding"]["location"].pop("table"), "namespace/database"), (lambda c, f: f["source"].update(schemaRegistryUrl="http://registry:8081"), "only to avro"), (lambda c, f: f["source"]["eventTime"].update(column="event_id"), "timestamp"), - (lambda c, f: c["builds"][0]["properties"].pop("sql"), "SELECT"), + (lambda c, f: c["builds"][0]["properties"].pop("stream"), "properties.stream"), + (lambda c, f: c["builds"][0]["properties"].update(sql=" "), "SELECT"), ( lambda c, f: c["exposes"][0]["binding"]["icebergConfig"].update(writeVersion=1), "version 2", @@ -221,7 +227,8 @@ def test_invalid_semantic_combinations_fail_before_emission(contract, mutate, ma def test_avro_source_emits_registry_format_without_json_options(contract): cfg = managed_config(contract["builds"][0]) - cfg["source"].update(format="avro-confluent", schemaRegistryUrl="https://registry.example") + _stream(contract)["format"] = "avro-confluent" + cfg["source"]["schemaRegistryUrl"] = "https://registry.example" (artifact,) = compile_contract(contract) assert "'avro-confluent.url' = 'https://registry.example'" in artifact.sql assert "json." not in artifact.sql @@ -277,3 +284,16 @@ def test_duplicate_writers_and_cross_build_state_prefixes_are_rejected(contract) contract["exposes"].append(target) with pytest.raises(ValueError, match="overlap between"): compile_contract(contract) + + +def test_event_time_needs_millisecond_precision(contract): + # Flink's rowtime takes precision 0 to 3; a stream timestamp defaults to 6. + stream = contract["builds"][0]["properties"]["stream"] + stream["schema"][-1]["type"] = "timestamp_ltz" + contract["exposes"][0]["contract"]["schema"][-1]["type"] = "timestamp_ltz" + with pytest.raises(ValueError, match="precision 3 or less"): + compile_contract(contract) + stream["metadata"] = [{"name": "kafka_ts", "from": "timestamp"}] + managed_config(contract["builds"][0])["source"]["eventTime"]["column"] = "kafka_ts" + (artifact,) = compile_contract(contract) + assert "WATERMARK FOR `kafka_ts`" in artifact.sql diff --git a/tests/streaming/test_contract_sql_gate.py b/tests/streaming/test_contract_sql_gate.py new file mode 100644 index 00000000..175b44bc --- /dev/null +++ b/tests/streaming/test_contract_sql_gate.py @@ -0,0 +1,145 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""A contract's own SQL runs only when the operator lists it. + +forge-cli writes the query from ``properties.mapping`` and that query reads only the +stream. A contract's ``properties.sql`` runs with the job's catalog identity and can read +any table that identity can, so apply refuses it unless the operator lists the product. +""" + +from __future__ import annotations + +import copy + +import pytest + +from fluid_build.streaming.compiler import validation_errors, validation_warnings +from fluid_build.streaming.policy import ( + CONTRACT_SQL_ALLOWLIST_ENV, + FlinkPolicyError, + contract_sql_entry, + refuse_unless_contract_sql_allowed, +) +from fluid_build.streaming.runtime import reconcile + + +def _props(contract): + return contract["builds"][0]["properties"] + + +def test_contract_sql_is_warned_about(contract): + _props(contract)["sql"] = "SELECT * FROM source_events" + (warning,) = validation_warnings(contract) + assert CONTRACT_SQL_ALLOWLIST_ENV in warning and "commerce.nested-events" in warning + + +def test_apply_refuses_contract_sql_the_operator_did_not_list(contract, monkeypatch): + _props(contract)["sql"] = "SELECT * FROM source_events" + monkeypatch.delenv(CONTRACT_SQL_ALLOWLIST_ENV, raising=False) + with pytest.raises(FlinkPolicyError, match="runs SQL written in the contract"): + reconcile(contract, dry_run=True) + build = contract["builds"][0] + for entry in ("commerce.nested-events", "commerce.nested-events/ingest-orders"): + refuse_unless_contract_sql_allowed( + contract["id"], build, {CONTRACT_SQL_ALLOWLIST_ENV: entry} + ) + with pytest.raises(FlinkPolicyError) as caught: + refuse_unless_contract_sql_allowed( + contract["id"], build, {CONTRACT_SQL_ALLOWLIST_ENV: "other.product"} + ) + pinned = contract_sql_entry(contract["id"], build) + assert f"other.product,{pinned}" in caught.value.fix + + +def test_a_pinned_listing_approves_the_reviewed_sql_only(contract): + build = contract["builds"][0] + _props(contract)["sql"] = "SELECT * FROM source_events" + pinned = contract_sql_entry(contract["id"], build) + assert pinned.startswith("commerce.nested-events/ingest-orders@sha256:") + refuse_unless_contract_sql_allowed(contract["id"], build, {CONTRACT_SQL_ALLOWLIST_ENV: pinned}) + # Whitespace around the SQL is not a change; anything else is. + _props(contract)["sql"] = "\n SELECT * FROM source_events \n" + refuse_unless_contract_sql_allowed(contract["id"], build, {CONTRACT_SQL_ALLOWLIST_ENV: pinned}) + _props(contract)["sql"] = "SELECT * FROM source_events UNION ALL SELECT * FROM lake.hr.pay" + with pytest.raises(FlinkPolicyError) as caught: + refuse_unless_contract_sql_allowed( + contract["id"], build, {CONTRACT_SQL_ALLOWLIST_ENV: pinned} + ) + assert "changed after it was listed" in caught.value.why + assert pinned not in caught.value.fix + assert contract_sql_entry(contract["id"], build) in caught.value.fix + + +def test_a_mapped_build_needs_no_listing(contract, monkeypatch): + monkeypatch.delenv(CONTRACT_SQL_ALLOWLIST_ENV, raising=False) + refuse_unless_contract_sql_allowed(contract["id"], contract["builds"][0]) + assert validation_warnings(contract) == [] + + +def test_a_second_build_with_the_same_id_cannot_vouch_for_the_first(contract, monkeypatch): + # The SQL build comes first; a clean build reuses its id. Builds were looked up + # by id at apply, so every check ran on the clean one and passed the other. + monkeypatch.delenv(CONTRACT_SQL_ALLOWLIST_ENV, raising=False) + first = contract["builds"][0] + second = copy.deepcopy(first) + first["properties"]["sql"] = "SELECT * FROM lake.secret_ns.payroll" + deployment = second["properties"]["parameters"]["flink"]["deployment"] + deployment["name"] = "commerce-orders-2" + for key in ("checkpointUri", "savepointUri", "haUri"): + deployment[key] = deployment[key].replace("/orders/", "/orders-2/") + second["outputs"] = ["orders2"] + expose = copy.deepcopy(contract["exposes"][0]) + expose["exposeId"] = "orders2" + expose["binding"]["location"]["table"] = "orders2" + contract["builds"].append(second) + contract["exposes"].append(expose) + with pytest.raises(ValueError, match="'ingest-orders' is used twice"): + reconcile(contract, dry_run=True) + assert any("used twice" in e for e in validation_errors(contract)) + # With distinct ids, the SQL build is judged as itself. + second["id"] = "ingest-orders-2" + with pytest.raises(FlinkPolicyError, match="'ingest-orders' runs SQL written in the contract"): + reconcile(contract, dry_run=True) + + +def test_nothing_is_applied_before_the_gate_and_a_listed_build_proceeds(contract, monkeypatch): + from unittest.mock import patch + + from fluid_build.streaming.compiler import compile_contract + from fluid_build.streaming.runtime import Kubectl + + _props(contract)["sql"] = "SELECT * FROM source_events" + monkeypatch.delenv(CONTRACT_SQL_ALLOWLIST_ENV, raising=False) + with patch.object(Kubectl, "get") as get, patch.object(Kubectl, "apply") as apply: + with pytest.raises(FlinkPolicyError): + reconcile(contract) + assert get.call_count == 0 and apply.call_count == 0 + (artifact,) = compile_contract(contract) + running = copy.deepcopy(artifact.resources[1]) + running["metadata"]["generation"] = 2 + running["status"] = { + "observedGeneration": 2, + "jobManagerDeploymentStatus": "READY", + "reconciliationStatus": {"state": "DEPLOYED"}, + "jobStatus": {"state": "RUNNING", "jobId": "a" * 32}, + } + entry = contract_sql_entry(contract["id"], contract["builds"][0]) + monkeypatch.setenv(CONTRACT_SQL_ALLOWLIST_ENV, entry) + with ( + patch.object(Kubectl, "get", side_effect=[None, None, running]), + patch.object(Kubectl, "apply") as apply, + ): + assert reconcile(contract)[0]["status"] == "running" + assert [c.args[0]["kind"] for c in apply.call_args_list] == ["ConfigMap", "FlinkDeployment"] diff --git a/tests/streaming/test_mapping.py b/tests/streaming/test_mapping.py new file mode 100644 index 00000000..cdff35e1 --- /dev/null +++ b/tests/streaming/test_mapping.py @@ -0,0 +1,814 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""Every Kafka record shape the managed lane takes, declared in FLUID and checked. + +Flat records, nested records kept nested, and nested records flattened (struct fields +lifted, arrays exploded, map entries looked up) or rebuilt into another nested shape; +the record key and Kafka metadata as columns. forge-cli writes every query, and an +output it cannot read with the declared type fails here, before anything deploys. +""" + +from __future__ import annotations + +import copy +import logging +from pathlib import Path + +import pytest +import yaml + +from fluid_build.schema_manager import FluidSchemaManager +from fluid_build.streaming.compiler import compile_contract, managed_config, validation_errors +from fluid_build.streaming.mapping import parse_path, show + + +def col(name, type_, **kw): + return {"name": name, "type": type_, **kw} + + +EVENT_ID = col("event_id", "string", required=True) +EVENT_TIME = col("event_time", "timestamp_ltz(3)") + + +def _props(contract): + return contract["builds"][0]["properties"] + + +def _stream(contract): + return _props(contract)["stream"] + + +def _output(contract, columns, mapping=None): + contract["exposes"][0]["contract"]["schema"] = columns + if mapping is not None: + _props(contract)["mapping"] = mapping + return contract + + +def _insert(contract): + (artifact,) = compile_contract(contract) + return artifact.statements[-1] + + +def _source_ddl(contract): + (artifact,) = compile_contract(contract) + return artifact.statements[3] + + +def _select(*lines, join=""): + body = ",\n ".join(lines) + return ( + "INSERT INTO `lake`.`commerce`.`orders`\nSELECT\n " + + body + + "\nFROM `source_events` AS `src`" + + join + ) + + +# ── shapes ────────────────────────────────────────────────────────────── + + +def test_a_nested_stream_lands_as_it_is_with_no_mapping(contract): + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "`src`.`customer` AS `customer`", + "`src`.`items` AS `items`", + "`src`.`attributes` AS `attributes`", + "`src`.`event_time` AS `event_time`", + ) + + +def test_a_flat_stream_lands_by_name_in_the_output_order(contract): + _stream(contract)["schema"] = [ + EVENT_ID, + col("customer_id", "string"), + col("amount", "decimal(18,2)"), + EVENT_TIME, + ] + _output(contract, [EVENT_TIME, col("amount", "decimal(18,2)"), EVENT_ID]) + assert _insert(contract) == _select( + "`src`.`event_time` AS `event_time`", + "`src`.`amount` AS `amount`", + "`src`.`event_id` AS `event_id`", + ) + + +def test_a_nested_event_flattens_to_one_row_per_item(contract): + _output( + contract, + [ + EVENT_ID, + col("customer_id", "string"), + col("country", "string"), + col("latitude", "double"), + col("channel", "string"), + col("sku", "string"), + col("price", "decimal(18,2)"), + EVENT_TIME, + ], + { + "explode": {"path": "items", "as": "item"}, + "columns": [ + {"name": "customer_id", "from": "customer.id"}, + {"name": "country", "from": "customer.address.country"}, + { + "name": "latitude", + "from": "customer.address.geo.location.coordinates.latitude", + }, + {"name": "channel", "from": "attributes['channel']"}, + {"name": "sku", "from": "item.sku"}, + {"name": "price", "from": "item.price"}, + ], + }, + ) + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "`src`.`customer`.`id` AS `customer_id`", + "`src`.`customer`.`address`.`country` AS `country`", + "`src`.`customer`.`address`.`geo`.`location`.`coordinates`.`latitude` AS `latitude`", + "`src`.`attributes`['channel'] AS `channel`", + "`elem`.`sku` AS `sku`", + "`elem`.`price` AS `price`", + "`src`.`event_time` AS `event_time`", + join="\nCROSS JOIN UNNEST(`src`.`items`) AS `elem` (`sku`, `price`)", + ) + assert validation_errors(contract) == [] + + +def test_a_scalar_array_explodes_and_its_element_is_read_by_its_name(contract): + _stream(contract)["schema"].append(col("tags", "array", element={"type": "string"})) + _output(contract, [EVENT_ID, col("tag", "string")], {"explode": {"path": "tags", "as": "tag"}}) + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "`elem`.`tag` AS `tag`", + join="\nCROSS JOIN UNNEST(`src`.`tags`) AS `elem` (`tag`)", + ) + + +def test_an_exploded_struct_element_can_stay_nested(contract): + item = col( + "item", + "struct", + fields=[col("sku", "string"), col("price", "decimal(18,2)")], + ) + _output(contract, [EVENT_ID, item], {"explode": {"path": "items", "as": "item"}}) + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "CAST(ROW(`elem`.`sku`, `elem`.`price`) AS ROW<`sku` STRING, `price` DECIMAL(18,2)>)" + " AS `item`", + join="\nCROSS JOIN UNNEST(`src`.`items`) AS `elem` (`sku`, `price`)", + ) + + +def test_a_flat_stream_can_be_written_as_a_nested_table(contract): + _stream(contract)["schema"] = [EVENT_ID, col("customer_id", "string"), col("country", "string")] + customer = col( + "customer", + "struct", + fields=[ + col("id", "string"), + col("address", "struct", fields=[col("country", "string")]), + ], + ) + _props(contract)["parameters"]["flink"]["source"].pop("eventTime") + _output( + contract, + [EVENT_ID, customer], + { + "columns": [ + {"name": "customer.id", "from": "customer_id"}, + {"name": "customer.address.country", "from": "country"}, + ] + }, + ) + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "CAST(ROW(`src`.`customer_id`, CAST(ROW(`src`.`country`) AS ROW<`country` STRING>))" + " AS ROW<`id` STRING, `address` ROW<`country` STRING>>) AS `customer`", + ) + + +def test_a_nested_subset_stays_null_where_its_source_struct_is_null(contract): + _output(contract, [EVENT_ID, col("customer", "struct", fields=[col("id", "string")])]) + assert _insert(contract) == _select( + "`src`.`event_id` AS `event_id`", + "CAST(CASE WHEN `src`.`customer` IS NULL THEN NULL " + "ELSE ROW(`src`.`customer`.`id`) END AS ROW<`id` STRING>) AS `customer`", + ) + + +def test_nested_fields_are_reordered_by_name_never_by_position(contract): + original = contract["exposes"][0]["contract"]["schema"][1] + swapped = col("customer", "struct", fields=list(reversed(original["fields"]))) + _output(contract, [EVENT_ID, swapped]) + insert = _insert(contract) + assert "ELSE ROW(`src`.`customer`.`address`, `src`.`customer`.`id`) END" in insert + assert insert.count("CASE WHEN `src`.`customer` IS NULL") == 1 + + +def test_a_type_is_widened_only_losslessly_and_said_so(contract): + _stream(contract)["schema"].append(col("quantity", "int")) + _output( + contract, + [EVENT_ID, col("quantity", "long"), col("event_time", "timestamp_ltz(6)")], + ) + insert = _insert(contract) + assert "CAST(`src`.`quantity` AS BIGINT) AS `quantity`" in insert + assert "CAST(`src`.`event_time` AS TIMESTAMP_LTZ(6)) AS `event_time`" in insert + + +# ── the output is checked before deploy ───────────────────────────────── + + +@pytest.mark.parametrize( + "columns,mapping,match", + [ + # One column the stream does not have: the SELECT cannot silently drop it. + ( + [EVENT_ID, col("channel", "string")], + None, + r"channel is not a column of properties.stream; map channel", + ), + ([col("event_id", "int", required=True)], None, r"event_id: string is not int"), + ( + [EVENT_ID, col("customer_id", "string", required=True)], + {"columns": [{"name": "customer_id", "from": "customer.id"}]}, + r"can be null, but the output requires it", + ), + ( + [EVENT_ID, col("channel", "string", required=True)], + {"columns": [{"name": "channel", "from": "attributes['channel']"}]}, + r"can be null, but the output requires it", + ), + ( + [EVENT_ID, col("customer", "struct", required=True, fields=[col("id", "string")])], + None, + r"required, but its fields are read from customer, which can be null", + ), + ( + [EVENT_ID, col("customer", "struct", fields=[col("id", "string", required=True)])], + None, + r"customer.id.*can be null, but the output requires it", + ), + ( + [ + EVENT_ID, + col( + "items", + "array", + element={"type": "struct", "fields": [col("sku", "string")]}, + ), + ], + None, + r"struct fields at \[\] are \['sku', 'price'\], not \['sku'\]", + ), + ( + [EVENT_ID, col("price", "decimal(10,2)")], + { + "explode": {"path": "items", "as": "i"}, + "columns": [{"name": "price", "from": "i.price"}], + }, + r"decimal\(18,2\) is not decimal\(10,2\)", + ), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "nope"}]}, + r"columns\[0\].from: nope is not a column of properties.stream", + ), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "customer.nope"}]}, + r"customer has no field nope", + ), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "event_id['k']"}]}, + r"event_id is string, not a map", + ), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "event_id.k"}]}, + r"event_id is string, not a struct", + ), + ( + [EVENT_ID], + {"columns": [{"name": "nope", "from": "event_id"}]}, + r"nope is not an output column or a struct field of one", + ), + ( + [ + EVENT_ID, + col( + "items", + "array", + element={ + "type": "struct", + "fields": [col("sku", "string"), col("price", "decimal(18,2)")], + }, + ), + ], + {"columns": [{"name": "items.sku", "from": "event_id"}]}, + r"items.sku is not an output column or a struct field of one", + ), + ( + [EVENT_ID, col("customer", "struct", fields=[col("id", "string")])], + { + "columns": [ + {"name": "customer", "from": "customer"}, + {"name": "customer.id", "from": "event_id"}, + ] + }, + r"customer.id is inside customer, mapped whole", + ), + ( + [EVENT_ID], + {"columns": [{"name": "event_id", "from": "event_id"}] * 2}, + r"event_id is mapped twice", + ), + ([EVENT_ID], {"explode": {"path": "customer", "as": "c"}}, r"customer is not an array"), + ( + [EVENT_ID], + {"explode": {"path": "items", "as": "event_id"}}, + r"event_id is already a column", + ), + ([EVENT_ID], {"explode": {"path": "items", "as": "item"}}, r"no output column reads item"), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "event_id", "cast": "int"}]}, + r"string cannot become int", + ), + ( + [EVENT_ID, col("x", "string")], + {"columns": [{"name": "x", "from": "customer", "cast": "struct"}]}, + r"only a scalar type", + ), + # Inside containers: an element field, a nullable element, a map value. + ( + [ + EVENT_ID, + col( + "items", + "array", + element={ + "type": "struct", + "fields": [col("sku", "string"), col("price", "string")], + }, + ), + ], + None, + r"decimal\(18,2\) at \[\]\.price is not string", + ), + ( + [ + EVENT_ID, + col( + "items", + "array", + element={ + "type": "struct", + "required": True, + "fields": [col("sku", "string"), col("price", "decimal(18,2)")], + }, + ), + ], + None, + r"a value at \[\] can be null", + ), + ( + [ + EVENT_ID, + col( + "attributes", + "map", + key={"type": "string", "required": True}, + value={"type": "int"}, + ), + ], + None, + r"string at \{value\} is not int", + ), + # Nullability survives a widening, a conversion and an exploded element. + ( + [EVENT_ID, col("n", "long", required=True)], + { + "columns": [ + { + "name": "n", + "from": "customer.address.geo.location.coordinates.latitude", + "cast": "double", + } + ] + }, + r"can be null, but the output requires it", + ), + ( + [EVENT_ID, col("sku", "string", required=True)], + { + "explode": {"path": "items", "as": "item"}, + "columns": [{"name": "sku", "from": "item.sku"}], + }, + r"can be null, but the output requires it", + ), + ([EVENT_ID], {"typo": True}, r"unsupported field\(s\) \['typo'\]"), + ([EVENT_ID], {"columns": []}, r"columns: expected a non-empty list"), + ], +) +def test_an_output_the_stream_cannot_fill_fails_before_deploy(contract, columns, mapping, match): + _output(contract, columns, mapping) + with pytest.raises(ValueError, match=match): + compile_contract(contract) + assert validation_errors(contract) + + +@pytest.mark.parametrize( + "path,match", + [ + ("event_id; DROP TABLE x", "expected a path"), + ("`event_id`", "expected a path"), + ("attributes['it''s']", "expected a path"), + ("attributes['a\\b']", "expected a path"), + ("event_id\n", "expected a path"), + ("customer..id", "expected a path"), + ("1st", "expected a path"), + ("attributes[channel]", "expected a path"), + ("attributes['" + "k" * 257 + "']", "expected a path"), + (".".join(["a"] * 65), "at most 64 steps"), + ], +) +def test_a_path_is_never_sql_and_a_bad_one_is_not_echoed(contract, path, match): + _output(contract, [EVENT_ID, col("x", "string")], {"columns": [{"name": "x", "from": path}]}) + with pytest.raises(ValueError, match=match) as caught: + compile_contract(contract) + assert path not in str(caught.value) + + +def test_paths_round_trip(): + for text in ("a", "a.b.c", "attributes['channel']", "headers['trace-id'].x"): + assert show(parse_path(text, "$")) == text + + +def test_contract_sql_is_bound_by_name(contract): + _props(contract)["sql"] = "SELECT * FROM source_events" + insert = _insert(contract) + assert insert.endswith( + "SELECT `event_id`, `customer`, `items`, `attributes`, `event_time`\n" + "FROM (\nSELECT * FROM source_events\n) AS `contract_sql`" + ) + + +def test_declaring_both_a_mapping_and_sql_is_refused(contract): + _props(contract)["mapping"] = {"columns": [{"name": "event_id", "from": "event_id"}]} + _props(contract)["sql"] = "SELECT * FROM source_events" + with pytest.raises(ValueError, match="not both"): + compile_contract(contract) + + +# ── key and Kafka metadata ────────────────────────────────────────────── + + +def test_a_raw_key_is_a_column_and_the_value_options_are_prefixed(contract): + _stream(contract)["key"] = {"format": "raw", "schema": [col("order_key", "string")]} + _output(contract, [EVENT_ID, col("order_key", "string")]) + ddl = _source_ddl(contract) + assert "`order_key` STRING," in ddl + for option in ( + "'key.fields' = 'order_key'", + "'key.format' = 'raw'", + "'value.fields-include' = 'EXCEPT_KEY'", + "'value.format' = 'json'", + "'value.json.ignore-parse-errors' = 'false'", + ): + assert option in ddl + assert "'format' =" not in ddl and "'json." not in ddl + assert "`src`.`order_key` AS `order_key`" in _insert(contract) + + +def test_a_record_key_that_repeats_a_value_field_takes_a_prefix(contract): + key = {"format": "json", "schema": [col("event_id", "string")]} + _stream(contract)["key"] = key + with pytest.raises(ValueError, match="declared twice.*set key.prefix"): + compile_contract(contract) + key["prefix"] = "key_" + _output(contract, [EVENT_ID, col("key_event_id", "string")]) + ddl = _source_ddl(contract) + assert "`key_event_id` STRING" in ddl + assert "'key.fields' = 'key_event_id'" in ddl + assert "'key.fields-prefix' = 'key_'" in ddl + assert "'key.json.fail-on-missing-field' = 'false'" in ddl + + +def test_an_avro_key_or_value_needs_the_registry(contract): + _stream(contract)["key"] = {"format": "avro-confluent", "schema": [col("k", "string")]} + with pytest.raises(ValueError, match="requires source.schemaRegistryUrl"): + compile_contract(contract) + managed_config(contract["builds"][0])["source"][ + "schemaRegistryUrl" + ] = "https://registry.example" + ddl = _source_ddl(contract) + assert "'key.avro-confluent.url' = 'https://registry.example'" in ddl + assert "'value.json.fail-on-missing-field' = 'false'" in ddl + + +@pytest.mark.parametrize( + "key,match", + [ + ({"format": "raw", "schema": [col("a", "string"), col("b", "string")]}, "one column"), + ({"format": "raw", "schema": [col("a", "int")]}, "string or binary"), + ({"format": "protobuf", "schema": [col("a", "string")]}, "key.format"), + ({"format": "raw", "schema": [col("a", "string")], "prefix": "1x"}, "key.prefix"), + ({"format": "raw", "schema": [col("a", "string")], "typo": 1}, "unsupported field"), + ], +) +def test_bad_keys_are_refused(contract, key, match): + _stream(contract)["key"] = key + with pytest.raises(ValueError, match=match): + compile_contract(contract) + + +def test_kafka_metadata_reads_as_typed_virtual_columns(contract): + _stream(contract)["metadata"] = [ + {"name": "kafka_offset", "from": "offset"}, + {"name": "kafka_partition", "from": "partition"}, + {"name": "kafka_ts", "from": "timestamp"}, + {"name": "kafka_headers", "from": "headers"}, + {"name": "kafka_epoch", "from": "leader-epoch"}, + ] + managed_config(contract["builds"][0])["source"]["eventTime"]["column"] = "kafka_ts" + _output( + contract, + [ + EVENT_ID, + col("kafka_offset", "long", required=True), + col("kafka_partition", "int"), + col("kafka_ts", "timestamp_ltz(3)", required=True), + col("trace_id", "string"), + col( + "kafka_headers", + "map", + key={"type": "string", "required": True}, + value={"type": "binary"}, + ), + col("kafka_epoch", "int"), + ], + {"columns": [{"name": "trace_id", "from": "kafka_headers['trace-id']", "cast": "string"}]}, + ) + ddl = _source_ddl(contract) + assert "`kafka_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL" in ddl + assert "`kafka_ts` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp' VIRTUAL" in ddl + assert "`kafka_headers` MAP NOT NULL METADATA FROM 'headers' VIRTUAL" in ddl + assert "`kafka_epoch` INT METADATA FROM 'leader-epoch' VIRTUAL" in ddl + assert "WATERMARK FOR `kafka_ts` AS `kafka_ts` - INTERVAL '5' SECOND" in ddl + insert = _insert(contract) + assert "CAST(`src`.`kafka_headers`['trace-id'] AS STRING) AS `trace_id`" in insert + assert "`src`.`kafka_offset` AS `kafka_offset`" in insert + + +@pytest.mark.parametrize( + "metadata,match", + [ + ([{"name": "m", "from": "key"}], "from: expected one of"), + ([{"name": "event_id", "from": "offset"}], "declared twice"), + ([{"from": "offset"}], "a column name is required"), + ([], "non-empty list"), + ], +) +def test_bad_metadata_is_refused(contract, metadata, match): + _stream(contract)["metadata"] = metadata + with pytest.raises(ValueError, match=match): + compile_contract(contract) + + +def test_leader_epoch_can_be_null(contract): + _stream(contract)["metadata"] = [{"name": "epoch", "from": "leader-epoch"}] + _output(contract, [EVENT_ID, col("epoch", "int", required=True)]) + with pytest.raises(ValueError, match="can be null"): + compile_contract(contract) + + +@pytest.mark.parametrize( + "mutate,match", + [ + (lambda s: s.update(kind="kinesis"), "only kafka"), + (lambda s: s.update(topic="orders;payments"), "one Kafka topic"), + (lambda s: s.update(topic="o'rders"), "one Kafka topic"), + (lambda s: s.update(typo=1), "unsupported field"), + (lambda s: s["schema"].append(col("bad`name", "string")), "Invalid SQL identifier"), + ], +) +def test_bad_streams_are_refused(contract, mutate, match): + mutate(_stream(contract)) + with pytest.raises(ValueError, match=match): + compile_contract(contract) + + +# ── the schema, the CLI and fragments ─────────────────────────────────── + + +def _validate(contract, tmp_path): + return FluidSchemaManager(cache_dir=tmp_path).validate_contract(contract, offline_only=True) + + +def test_the_example_validates(contract, tmp_path): + result = _validate(contract, tmp_path) + assert result.is_valid, result.errors + + +@pytest.mark.parametrize( + "mutate,match", + [ + # A key the column definition does not have is refused, not silently dropped. + (lambda c: _stream(c)["schema"][0].update(metadata="offset"), "metadata"), + (lambda c: _props(c).pop("stream"), "'sql' is a required property"), + ( + lambda c: _props(c).update(mapping={"columns": [{"name": "a", "from": "b"}]}) + or _props(c).pop("stream"), + "'stream' is a dependency of 'mapping'", + ), + ( + lambda c: _props(c).update(mapping={"columns": [{"name": "a", "from": "b; x"}]}), + "does not match", + ), + (lambda c: _stream(c).update(format="protobuf"), "protobuf"), + (lambda c: c.update(fluidVersion="0.7.6"), "stream"), + ], +) +def test_the_schema_checks_streams_and_mappings(contract, tmp_path, mutate, match): + mutate(contract) + result = _validate(contract, tmp_path) + assert not result.is_valid + assert any(match in str(error) for error in result.errors), result.errors + + +def _write(path: Path, document) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8") + return path + + +def _fragment_project(tmp_path, contract): + """The example, with its stream and a mapping kept in fragments, as a team would.""" + root = tmp_path / "product" + stream = copy.deepcopy(_stream(contract)) + value = _write(root / "fragments/streams/orders.value.yaml", stream.pop("schema")) + stream["schema"] = {"$ref": "./orders.value.yaml"} + _write(root / "fragments/streams/orders.yaml", stream) + mapping = { + "explode": {"path": "items", "as": "item"}, + "columns": [{"name": "sku", "from": "item.sku"}, {"name": "price", "from": "item.price"}], + } + _write(root / "fragments/mappings/order-items.yaml", mapping) + _props(contract)["stream"] = {"$ref": "./fragments/streams/orders.yaml"} + _props(contract)["mapping"] = {"$ref": "./fragments/mappings/order-items.yaml"} + contract["exposes"][0]["contract"]["schema"] = [ + EVENT_ID, + col("sku", "string"), + col("price", "decimal(18,2)"), + EVENT_TIME, + ] + return _write(root / "contract.fluid.yaml", contract), value, mapping + + +def test_the_stream_and_mapping_work_from_fragments(contract, tmp_path, capsys): + from fluid_build._contract_loader import load_contract_with_overlay + from fluid_build.cli import main + + inline = copy.deepcopy(contract) + path, value, mapping = _fragment_project(tmp_path, contract) + _props(inline)["mapping"] = mapping + inline["exposes"] = copy.deepcopy(contract["exposes"]) + loaded = load_contract_with_overlay(str(path), None, logging.getLogger("test")) + assert compile_contract(loaded) == compile_contract(inline) + + cache = ["--cache-dir", str(tmp_path / "cache")] + assert main(["validate", *cache, str(path)]) == 0 + root = path.parent + for kind, fragment in ( + ("streamSource", root / "fragments/streams/orders.yaml"), + ("recordMapping", root / "fragments/mappings/order-items.yaml"), + ("schema", value), + ): + assert main(["validate", *cache, "--fragment", kind, str(fragment)]) == 0, kind + _write(root / "fragments/mappings/order-items.yaml", {**mapping, "typo": 1}) + assert ( + main( + [ + "validate", + *cache, + "--fragment", + "recordMapping", + str(root / "fragments/mappings/order-items.yaml"), + ] + ) + == 1 + ) + capsys.readouterr() + + +def test_a_widened_column_keeps_its_nullability(contract): + _stream(contract)["schema"].append(col("quantity", "int")) + _output(contract, [EVENT_ID, col("quantity", "long", required=True)]) + with pytest.raises(ValueError, match="can be null, but the output requires it"): + compile_contract(contract) + _stream(contract)["schema"][-1]["required"] = True + assert "CAST(`src`.`quantity` AS BIGINT) AS `quantity`" in _insert(contract) + + +def test_a_required_exploded_field_needs_a_required_element(contract): + items = _stream(contract)["schema"][2] + items["element"]["fields"][0]["required"] = True + _output( + contract, + [EVENT_ID, col("sku", "string", required=True)], + { + "explode": {"path": "items", "as": "item"}, + "columns": [{"name": "sku", "from": "item.sku"}], + }, + ) + with pytest.raises(ValueError, match="can be null"): + compile_contract(contract) + items["element"]["required"] = True + assert "`elem`.`sku` AS `sku`" in _insert(contract) + + +def test_a_required_field_of_an_optional_struct_is_required_where_the_struct_is_present( + contract, +): + # The rebuilt struct is null where the customer is; where the customer is there, + # its required id is too. A subset, a reorder and a widening all keep it required. + customer = _stream(contract)["schema"][1] + customer["fields"][0]["required"] = True + customer["fields"].append(col("visits", "int", required=True)) + _output( + contract, + [ + EVENT_ID, + col( + "customer", + "struct", + fields=[col("visits", "long", required=True), col("id", "string", required=True)], + ), + ], + ) + insert = _insert(contract) + assert ( + "CAST(CASE WHEN `src`.`customer` IS NULL THEN NULL ELSE " + "ROW(CAST(`src`.`customer`.`visits` AS BIGINT), `src`.`customer`.`id`) END " + "AS ROW<`visits` BIGINT NOT NULL, `id` STRING NOT NULL>) AS `customer`" in insert + ) + # The struct itself still can be null, so it cannot be declared required. + contract["exposes"][0]["contract"]["schema"][1]["required"] = True + with pytest.raises(ValueError, match="required, but its fields are read from customer"): + compile_contract(contract) + + +@pytest.mark.parametrize( + "mutate,match", + [ + (lambda s: s.update(topic="."), "one Kafka topic"), + (lambda s: s.update(topic=".."), "one Kafka topic"), + (lambda s: s.update(key="raw"), "key: expected an object"), + (lambda s: s.update(key=[1]), "key: expected an object"), + ( + lambda s: s.update(key={"format": "raw", "schema": [col("k", "string")], "prefix": ""}), + "key.prefix", + ), + (lambda s: s.update(metadata=[{"name": "m", "from": ["offset"]}]), "from: expected one of"), + ], +) +def test_malformed_streams_fail_with_a_message_not_a_crash(contract, mutate, match): + mutate(_stream(contract)) + with pytest.raises(ValueError, match=match): + compile_contract(contract) + assert validation_errors(contract) + + +def test_warnings_never_crash_and_name_a_mapping_nothing_compiles(contract): + from fluid_build.streaming.compiler import validation_warnings + + assert validation_warnings({**contract, "builds": 5}) == [] + unmanaged = copy.deepcopy(contract) + _props(unmanaged)["parameters"].pop("flink") + _props(unmanaged)["sql"] = "SELECT 1" + _props(unmanaged)["mapping"] = {"columns": [{"name": "event_id", "from": "event_id"}]} + (warning,) = validation_warnings(unmanaged) + assert "properties.mapping is compiled only by the managed Flink lane" in warning + + +def test_an_unmanaged_build_with_a_stream_still_needs_its_sql(contract, tmp_path): + _props(contract)["parameters"].pop("flink") + result = _validate(contract, tmp_path) + assert any("'sql' is a required property" in str(e) for e in result.errors), result.errors + _props(contract)["sql"] = "SELECT * FROM orders" + assert _validate(contract, tmp_path).is_valid diff --git a/tests/streaming/test_runtime.py b/tests/streaming/test_runtime.py index 1b59a24d..65961a30 100644 --- a/tests/streaming/test_runtime.py +++ b/tests/streaming/test_runtime.py @@ -264,6 +264,7 @@ def test_catalog_auth_is_resolved_locally_and_sent_only_as_headers( contract, monkeypatch, auth_type ): contract["exposes"][0]["contract"]["schema"] = [{"name": "id", "type": "long"}] + contract["builds"][0]["properties"]["stream"]["schema"].append({"name": "id", "type": "long"}) contract["exposes"][0]["binding"]["location"]["uri"] = "https://catalog.example" monkeypatch.setenv(HOST_ALLOWLIST_ENV, "catalog.example,identity.example") cfg = contract["builds"][0]["properties"]["parameters"]["flink"] @@ -511,6 +512,7 @@ def test_successful_checkpoint_and_catalog_evidence(contract): def test_missing_or_stale_commit_refused(contract, snapshots, current): # Keep one scalar to isolate snapshot evidence from recursive schema checks. contract["exposes"][0]["contract"]["schema"] = [{"name": "id", "type": "long"}] + contract["builds"][0]["properties"]["stream"]["schema"].append({"name": "id", "type": "long"}) (artifact,) = compile_contract(contract) metadata = { "current-schema-id": 0, diff --git a/tests/streaming/test_standards.py b/tests/streaming/test_standards.py index 583b583a..24369723 100644 --- a/tests/streaming/test_standards.py +++ b/tests/streaming/test_standards.py @@ -48,10 +48,18 @@ def test_recursive_odcs_native_shape_and_lossless_extension(contract): assert prop["customProperties"][0]["property"] == "fluid.recursiveType" +def _schemas(contract): + """The output and the stream it is read from: a changed field stays readable.""" + return ( + contract["exposes"][0]["contract"]["schema"], + contract["builds"][0]["properties"]["stream"]["schema"], + ) + + def test_recursive_field_diff_in_canonical_contract(contract): new = copy.deepcopy(contract) - items = new["exposes"][0]["contract"]["schema"][2]["element"] - items["fields"][1]["type"] = "string" + for schema in _schemas(new): + schema[2]["element"]["fields"][1]["type"] = "string" report = compare_contracts(contract, new) assert report.has_breaking assert any("items.element.fields.price.type" in change.path for change in report.breaking) @@ -59,20 +67,22 @@ def test_recursive_field_diff_in_canonical_contract(contract): def test_nested_required_field_addition_is_breaking(contract): new = copy.deepcopy(contract) - new["exposes"][0]["contract"]["schema"][1]["fields"].append( - { - "name": "new_required", - "type": "string", - "required": True, - } - ) + for schema in _schemas(new): + schema[1]["fields"].append( + { + "name": "new_required", + "type": "string", + "required": True, + } + ) report = compare_contracts(contract, new) assert report.has_breaking def test_array_element_nullability_tightening_is_breaking(contract): new = copy.deepcopy(contract) - new["exposes"][0]["contract"]["schema"][2]["element"]["required"] = True + for schema in _schemas(new): + schema[2]["element"]["required"] = True report = compare_contracts(contract, new) assert report.has_breaking diff --git a/tests/streaming/test_stream_diff.py b/tests/streaming/test_stream_diff.py new file mode 100644 index 00000000..ca7326e6 --- /dev/null +++ b/tests/streaming/test_stream_diff.py @@ -0,0 +1,222 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""``fluid diff`` on what a streaming build reads: a reader schema, and its mapping.""" + +from __future__ import annotations + +import copy + +import pytest + +from fluid_build.api.changelog import compare_contracts +from fluid_build.api.changelog_stream_rules import diff_streams +from fluid_build.streaming.compiler import validation_errors + + +@pytest.fixture +def baseline(contract): + """The example, reading two fields no output uses, so either can change.""" + stream = contract["builds"][0]["properties"]["stream"] + stream["schema"] += [ + {"name": "note", "type": "string"}, + {"name": "quantity", "type": "int"}, + ] + return contract + + +def _stream(contract): + return contract["builds"][0]["properties"]["stream"] + + +def _registry(contract, url): + """The registry the avro-confluent format needs.""" + contract["builds"][0]["properties"]["parameters"]["flink"]["source"]["schemaRegistryUrl"] = url + + +def _kinds(baseline, new): + assert validation_errors(new) == [] + report = compare_contracts(baseline, new) + found = {(c.kind, c.severity) for c in report.breaking + report.non_breaking + report.info} + return {k for k in found if k[0].startswith("stream_")} + + +def test_an_unchanged_stream_reports_nothing(baseline): + assert diff_streams(baseline, copy.deepcopy(baseline)) == [] + + +@pytest.mark.parametrize( + "mutate,expected", + [ + (lambda s, c: s.update(topic="orders-v2"), ("stream_topic_changed", "breaking")), + ( + lambda s, c: s.update(format="avro-confluent") + or _registry(c, "https://registry.example"), + ("stream_format_changed", "breaking"), + ), + # The job stops reading a field: an output still reading it would fail validate. + (lambda s, c: s["schema"].pop(), ("stream_field_removed", "non_breaking")), + ( + lambda s, c: s["schema"].append({"name": "channel", "type": "string"}), + ("stream_field_added", "non_breaking"), + ), + ( + lambda s, c: s["schema"].append( + {"name": "channel", "type": "string", "required": True} + ), + ("stream_field_added", "breaking"), + ), + ( + lambda s, c: s["schema"][-1].update(type="long"), + ("stream_column_type_widened", "non_breaking"), + ), + ( + lambda s, c: s["schema"][-2].update(type="int"), + ("stream_column_type_changed", "breaking"), + ), + ( + lambda s, c: s["schema"][-2].update(required=True), + ("stream_nullable_tightened", "breaking"), + ), + ( + lambda s, c: s["schema"][1]["fields"].append({"name": "tier", "type": "string"}), + ("stream_field_added", "non_breaking"), + ), + ( + lambda s, c: s.update(metadata=[{"name": "kafka_offset", "from": "offset"}]), + None, + ), + ], +) +def test_reader_schema_rules(baseline, mutate, expected): + new = copy.deepcopy(baseline) + mutate(_stream(new), new) + kinds = _kinds(baseline, new) + assert kinds == ({expected} if expected else set()), kinds + + +def test_a_nested_change_names_its_path(baseline): + new = copy.deepcopy(baseline) + _stream(new)["schema"][1]["fields"].append({"name": "tier", "type": "string"}) + (change,) = diff_streams(baseline, new) + assert change.path == "builds[0].properties.stream.schema.customer.fields.tier" + assert "the job now reads 'customer.tier'" in change.description + + +def test_key_metadata_explode_and_sources(baseline): + _stream(baseline)["key"] = {"format": "raw", "schema": [{"name": "k", "type": "string"}]} + _stream(baseline)["metadata"] = [{"name": "kafka_meta", "from": "offset"}] + new = copy.deepcopy(baseline) + _stream(new)["key"]["format"] = "json" + _stream(new)["metadata"][0]["from"] = "partition" + assert _kinds(baseline, new) == { + ("stream_key_format_changed", "breaking"), + ("stream_metadata_source_changed", "breaking"), + } + + rows = copy.deepcopy(baseline) + rows["builds"][0]["properties"]["mapping"] = { + "explode": {"path": "items", "as": "item"}, + "columns": [{"name": "sku", "from": "item.sku"}, {"name": "label", "from": "note"}], + } + rows["exposes"][0]["contract"]["schema"] = [ + {"name": "event_id", "type": "string", "required": True}, + {"name": "sku", "type": "string"}, + {"name": "label", "type": "string"}, + ] + assert ("stream_explode_changed", "breaking") in _kinds(baseline, rows) + + moved = copy.deepcopy(rows) + moved["builds"][0]["properties"]["mapping"]["columns"][1]["from"] = "customer.id" + assert _kinds(rows, moved) == {("stream_column_source_changed", "non_breaking")} + + +def test_a_renamed_build_is_a_new_build(baseline): + other = copy.deepcopy(baseline) + other["builds"][0]["id"] = "renamed" + assert diff_streams(baseline, other) == [] + # Renaming it is no way to slip its own SQL past review. + other["builds"][0]["properties"]["sql"] = "SELECT * FROM lake.hr.pay" + (change,) = diff_streams(baseline, other) + assert (change.kind, change.severity) == ("stream_contract_sql_added", "breaking") + assert "new build 'renamed'" in change.description + + +def test_a_build_without_a_stream_on_either_side_is_left_alone(baseline): + other = copy.deepcopy(baseline) + for contract in (baseline, other): + contract["builds"][0]["properties"].pop("stream") + other["builds"][0]["properties"]["sql"] = "SELECT 2" + assert diff_streams(baseline, other) == [] + + +def test_a_stream_declared_or_dropped_is_reported(baseline): + without = copy.deepcopy(baseline) + without["builds"][0]["properties"].pop("stream") + (dropped,) = diff_streams(baseline, without) + assert (dropped.kind, dropped.severity) == ("stream_dropped", "info") + (declared,) = diff_streams(without, baseline) + assert (declared.kind, declared.severity) == ("stream_declared", "info") + + +def test_key_schema_explode_removal_and_casts_are_reported(baseline): + _stream(baseline)["key"] = {"format": "raw", "schema": [{"name": "k", "type": "string"}]} + keyed = copy.deepcopy(baseline) + _stream(keyed)["key"]["schema"][0]["type"] = "binary" + assert _kinds(baseline, keyed) == {("stream_column_type_changed", "breaking")} + + rows = copy.deepcopy(baseline) + rows["builds"][0]["properties"]["mapping"] = { + "explode": {"path": "items", "as": "item"}, + "columns": [{"name": "sku", "from": "item.sku"}], + } + rows["exposes"][0]["contract"]["schema"] = [ + {"name": "event_id", "type": "string", "required": True}, + {"name": "sku", "type": "string"}, + ] + flat = copy.deepcopy(rows) + flat["builds"][0]["properties"].pop("mapping") + flat["exposes"][0]["contract"]["schema"] = [ + {"name": "event_id", "type": "string", "required": True} + ] + assert ("stream_explode_changed", "breaking") in _kinds(rows, flat) + + cast = copy.deepcopy(baseline) + _stream(cast)["metadata"] = [{"name": "kafka_headers", "from": "headers"}] + cast["builds"][0]["properties"]["mapping"] = { + "columns": [{"name": "trace", "from": "kafka_headers['trace']"}] + } + cast["exposes"][0]["contract"]["schema"].append({"name": "trace", "type": "binary"}) + converted = copy.deepcopy(cast) + converted["builds"][0]["properties"]["mapping"]["columns"][0]["cast"] = "string" + converted["exposes"][0]["contract"]["schema"][-1]["type"] = "string" + (source,) = [ + c for c in diff_streams(cast, converted) if c.kind.startswith("stream_column_source") + ] + assert source.after == "kafka_headers['trace'] as string" + + +def test_a_build_that_gains_its_stream_and_new_sql_in_one_edit_is_reported(baseline): + # The base branch's managed shape had no stream; migrating to it is one edit, + # and so is switching to SQL that reads other tables in the same edit. + old = copy.deepcopy(baseline) + old["builds"][0]["properties"].pop("stream") + old["builds"][0]["properties"]["sql"] = "SELECT * FROM source_events" + new = copy.deepcopy(baseline) + new["builds"][0]["properties"]["sql"] = "SELECT * FROM lake.hr.salaries" + kinds = {(c.kind, c.severity) for c in diff_streams(old, new)} + assert kinds == {("stream_contract_sql_changed", "breaking"), ("stream_declared", "info")} + old["builds"][0]["properties"].pop("sql") + kinds = {(c.kind, c.severity) for c in diff_streams(old, new)} + assert kinds == {("stream_contract_sql_added", "breaking"), ("stream_declared", "info")} diff --git a/tests/streaming/test_stream_lineage.py b/tests/streaming/test_stream_lineage.py new file mode 100644 index 00000000..c83c9ef8 --- /dev/null +++ b/tests/streaming/test_stream_lineage.py @@ -0,0 +1,82 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +"""ODCS export names where each output column of a stream mapping is read from.""" + +from __future__ import annotations + +from fluid_build.providers.odcs import OdcsProvider + + +def col(name, type_, **kw): + return {"name": name, "type": type_, **kw} + + +EVENT_ID = col("event_id", "string", required=True) +EVENT_TIME = col("event_time", "timestamp_ltz(3)") + + +def _props(contract): + return contract["builds"][0]["properties"] + + +def _output(contract, columns, mapping=None): + contract["exposes"][0]["contract"]["schema"] = columns + if mapping is not None: + _props(contract)["mapping"] = mapping + return contract + + +def test_odcs_export_carries_each_columns_source(contract): + _output( + contract, + [ + EVENT_ID, + col("country", "string"), + col("customer", "struct", fields=[col("id", "string")]), + col("sku", "string"), + EVENT_TIME, + ], + { + "explode": {"path": "items", "as": "item"}, + "columns": [ + {"name": "country", "from": "customer.address.country"}, + {"name": "customer.id", "from": "customer.id"}, + {"name": "sku", "from": "item.sku"}, + ], + }, + ) + provider = OdcsProvider() + ((expose_id, odcs),) = provider.render_all_ports(contract) + (schema,) = odcs["schema"] + assert schema["dataGranularityDescription"] == ( + "One row per element of items in each record of topic orders" + ) + lineage = { + p["name"]: (p.get("transformSourceObjects"), p.get("transformLogic")) + for p in schema["properties"] + } + assert lineage["event_id"] == (["orders"], "event_id") + assert lineage["country"] == (["orders"], "customer.address.country") + assert lineage["sku"] == (["orders"], "item.sku") + assert lineage["customer"] == (None, None) # built field by field: its field says + (customer_id,) = next(p for p in schema["properties"] if p["name"] == "customer")["properties"] + assert customer_id["transformLogic"] == "customer.id" + assert provider.roundtrip_check(odcs)["equal"] + + +def test_contract_sql_has_no_lineage_to_report(contract): + _props(contract)["sql"] = "SELECT * FROM source_events" + ((_, odcs),) = OdcsProvider().render_all_ports(contract) + assert not any("transformLogic" in p for p in odcs["schema"][0]["properties"])