feat(streaming): declare every Kafka record shape in FLUID, checked before deploy

.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"

.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/

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 `<product>/<build>@sha256:<digest>`, which the refusal prints. That
+entry approves the SQL that was reviewed and nothing it is later edited into.
+`<product>/<build>` and `<product>` 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.

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

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

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: .

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)

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