GTFS-Realtime: Source snapshots and declared SQL transforms

One gtfs_realtime Source represents one protobuf endpoint. Its one stream, feed_snapshots, appends a complete observation to Bronze on each successful poll. SQL Pipeline declarations turn those observations into typed history, current state and Static correlations.

Use the same explicit feed_key on an operator's Static and Realtime Sources. Set endpoint_key when you need a stable endpoint identity independent of a Source UUID. It defaults to the Source UUID. Endpoint names do not select entity types: every descriptor-supported entity in the response is retained.

Configure and inspect the Source

Declare a Source using the normal apply format:

apiVersion: frank.platform/v1
kind: Source
metadata:
  name: operator-realtime
spec:
  name: operator-realtime
  pattern_id: gtfs_realtime
  source_config:
    feed_url: https://provider.example/gtfs-realtime.pb
    feed_key: operator
    endpoint_key: realtime
    auth_type: none
    request_timeout_seconds: 30
    max_response_bytes: 33554432
  schedule_type: manual
  streams:
    - name: feed_snapshots
      sync_mode: full_refresh
      write_disposition: append
      primary_key_path: [snapshot_id]

Replace the example URL with the provider's actual endpoint. Shared Source authentication supports the existing API key, bearer, basic and OAuth forms; credential storage follows the deployment's configured credential mode. HTTP acquisition validates public destinations, pins the resolved address, verifies TLS and bounds response size and request time. Provider URLs are never fetched inside the SQL transforms.

Use frankctl to store the declaration, discover the stream and sync it:

frankctl pipelines apply -f source.yaml --dry-run
frankctl pipelines apply -f source.yaml --dry-run-server
frankctl pipelines apply -f source.yaml
frankctl sources discover <source-id>
frankctl sources streams refresh-schema <source-id> --from-discovery
frankctl sources sync <source-id>
frankctl sources streams list <source-id>
frankctl sources history <source-id>

Read the physical Bronze table from the Source's resulting stream metadata. The API's semantic bronze.namespace.table path and Trino's configured catalog.namespace.table path identify the same table through different interfaces; use the configured Trino catalog in SQL.

What a snapshot retains

protobuf_payload is an Iceberg binary column containing the exact response bytes. decoded_json is canonical JSON decoded with MobilityData's official GTFS-Realtime bindings, pinned to version 2.2.0. Unknown protobuf fields remain recoverable from the original binary even when absent from decoded JSON.

Each row also contains the payload SHA-256 and byte size, entity and variant counts, header version/incrementality/timestamp, fetch start and completion times, feed_key, endpoint_key, and standard Source/extraction lineage. A valid empty feed produces a row with entity_count = 0.

The row's snapshot_id is the extraction UUID. An Iceberg snapshot ID is a separate table-commit identifier. Keep Iceberg IDs as decimal strings in JSON, JavaScript, SDK and CLI output; never convert them to JavaScript numbers. A retry of the same committed extraction does not append twice. A later poll of identical bytes is a distinct observation and appends another row.

Declare history and current transforms

The gtfs_realtime_history and gtfs_realtime_current catalog patterns render ordinary Trino SQL. Their parameters are source_table and projection. Supported projections are:

• vehicle_positions, trip_updates, stop_time_updates;
• alerts, alert_active_periods, alert_informed_entities, alert_translations;
• shapes, stops, trip_modifications;
• diagnostics for unsupported current-state incrementality.

For example, declare the Source ID and actual Bronze table at Pipeline level, then select that input explicitly for both steps:

apiVersion: frank.platform/v1
kind: Pipeline
metadata:
  name: operator-realtime-projections
spec:
  name: operator-realtime-projections
  source_ids: [<source-id>]
  source_tables: [lakehouse.operator_bronze.feed_snapshots]
  schedule_config: {type: manual}
  steps:
  - name: vehicle_history
    kind: catalog_pattern
    depends_on: []
    sources: [lakehouse.operator_bronze.feed_snapshots]
    config:
      output_layer: silver
      materialization: incremental
      merge_keys: [record_id]
      incremental_mode: watermark
      cursor_field: _extracted_at
      tiebreaker_field: _record_id
    pattern_id: gtfs_realtime_history
    pattern_params:
      source_table: lakehouse.operator_bronze.feed_snapshots
      projection: vehicle_positions
  - name: vehicle_current
    kind: catalog_pattern
    depends_on: []
    sources: [lakehouse.operator_bronze.feed_snapshots]
    config:
      output_layer: silver
      materialization: table
    pattern_id: gtfs_realtime_current
    pattern_params:
      source_table: lakehouse.operator_bronze.feed_snapshots
      projection: vehicle_positions

Replace the example table with the actual Bronze table. Both steps consume Bronze changes directly. Do not make current-state updates depend only on history-table changes: an empty feed must clear current state even though it adds no history entities.

History is insert-only and idempotent by snapshot/entity/child identity. Current state selects the latest FULL_DATASET snapshot per endpoint before expanding entities. Missing entities disappear, and an empty complete feed clears the endpoint. DIFFERENTIAL and unknown incrementality remain in Bronze and produce explicit diagnostics; they do not change current state.

Every projection preserves Source, extraction, snapshot, endpoint, feed, entity ordinal and Transform identity. Child projections retain every repeated child and its ordinal. Complete entity and child JSON remain available alongside typed columns, including fields not promoted into individual scalar columns.

For scheduled feeds, declare incremental_mode: watermark on history steps so each successful TransformRun reads only Bronze rows after the preceding successful run's cursor. Frank rewrites the declared qualified input relation to its generated delta CTE at execution time; the stored artifact and lineage continue to name the physical Bronze table. Keep current-state steps in full_refresh mode because they must retain the preceding complete state when the newest snapshot is DIFFERENTIAL.

Correlate with Static using declared SQL

Author correlation as kind: custom_sql in the Pipeline declaration. Use tenant- and feed-scoped LEFT JOINs on raw trip, route and stop IDs. Preserve the Realtime columns and Static Source/extraction identifiers in the result. Expose match flags so unmatched references can be counted and sampled.

An operator's trip 123 must never match another operator's trip 123. Likewise, normalization into existing downstream keys must not silently collapse distinct raw IDs such as 00123 and 123.

GTFS Static discovery and extraction preserve ID columns as their original CSV text, including leading zeros, large numeric-looking IDs and literal NA values. Optional empty IDs remain empty. Numeric fields such as direction_id, coordinates and distances retain numeric inference. This reader contract applies through the GTFS connector profile to existing and new Sources; it does not reconstruct IDs already changed in Bronze or migrate existing numeric columns. Saved numeric stream schemas still coerce newly read ID text back to numbers downstream. Use a new proof Source with fresh discovery or an explicitly reviewed schema and data migration for those historical tables.

For SQL that exposes a compatibility history, declare config.materialization: incremental, config.merge_keys: [external_id] and config.write_mode: append_only. Frank reads this policy from the current tenant-owned Pipeline version and performs an insert-only MERGE. An ordinary incremental SQL declaration defaults to write_mode: upsert. Catalog patterns own their write contract through their pattern definition.

The authoring loop stores, tests, executes and reads back the same declaration:

frankctl pipelines apply -f pipeline.yaml --dry-run-server
frankctl pipelines apply -f pipeline.yaml
frankctl pipelines validate <pipeline-id>
frankctl pipelines apply -f pipeline.yaml --wait
frankctl pipelines get <pipeline-id> --include-version
frankctl pipelines export <pipeline-id> -o yaml
frankctl transforms runs <transform-id>
frankctl datasets preview <output-table> --limit 10

Scheduling and evidence

Source polling uses normal Temporal interval schedules with overlap policy SKIP; Pipeline scheduling uses the existing Dagster machinery. Select a cadence from measured fetch-to-Bronze and projection-query latency for the deployment. A configured 10–60 second interval alone is not evidence that workers sustain it.

Schedule readback exposes executed action counts, overlap skips and missed catchup counts. running_actions contains exact workflow and run identities; null means observation is unavailable, while an empty list means no running actions were observed. Action timestamps remain in recent_actions. Capture successive observations because the recent-action list is bounded. Compare each action's exact workflow identity with SourceSync details, extraction UUIDs, Bronze rows and table-commit evidence. Correct polling behavior and available worker capacity are separate acceptance checks.

Before removing a temporary proof Source, pause its schedule and verify that Temporal has no running actions and every observed scheduled execution has a matching terminal SyncRun. Empty SyncRun history alone is insufficient: a scheduled workflow can start before its history row is created.

The repository's durable journey is tests/e2e-live/specs/10_sources/gtfs_realtime_lifecycle.spec.ts. It uses the published CLI to store declarations, Playwright for authenticated UI checks, and exact-version Trino reads during a cohort of at least twenty scheduled polls. Configure GTFS_E2E_INTERVAL_SECONDS between 10 and 60. Each poll must be observed queryable within that interval from both its scheduled time and HTTP fetch start. First successful query completion is an upper bound on visibility, not an inferred availability timestamp. Forecasts and actual actions account for due slots; a passing cohort requires zero missed-catchup invocations, so every remaining due slot reconciles to an intentional overlap skip. The API does not expose individual skip-event timestamps.

The cohort observes Bronze commits in a separate sequential query loop while collecting schedule and terminal responses. Slow control-plane reads therefore do not suspend ongoing exact-version queries after the Bronze table is found. Query failures still fail the cohort, and cleanup waits for the observer to stop. The initial table lookup and each query retain their transport bounds; this separation does not establish worker capacity or excuse API failures.

The journey also reads Bronze and Silver through the transform-author SDK installed in the selected deployed API image. Supply an explicit Kubernetes context, namespace, pod and container using GTFS_E2E_KUBE_CONTEXT, GTFS_E2E_KUBE_NAMESPACE, GTFS_E2E_SDK_POD and GTFS_E2E_SDK_CONTAINER. GTFS_E2E_API_IMAGE_DIGEST must identify the accepted release image; GTFS_E2E_SDK_CONTEXT_SHA256 and GTFS_E2E_SDK_RESULT_SHA256 must match its pinned shared-utils source. The helper installs nothing and uses the image's existing Trino configuration. It compares full row identities and exact snapshot strings with API/CLI/direct-query evidence. Context IDs come from the verified run response; this checks SDK context preservation, rather than claiming the transform-author SDK discovers control-plane resources.

Run this proof only with an explicit staging target, matching Static Source and table identities, valid CLI profile, and Playwright SSO state or credentials. Retain its run and table receipts before cleanup. Deleting a Source or Transform definition does not prove physical Iceberg data deletion; temporary data needs separate, verified cleanup through the catalog's supported API after writers have drained. Existing CIRA and historical data are never cleanup targets.

Existing BackingDataset rematerialization

For existing ontology consumers, run tests/e2e-live/specs/40_ontology/backing_dataset_rematerialization.spec.ts. This read-only proof observes scheduled full-refresh SQL transforms and matches twenty distinct output snapshots per configured BackingDataset to successful ontology writes using the same SQL artifact and mapping fingerprint. It also checks the existing dataset ID, mappings, final compatible health, published CLI response and authenticated BackingDataset detail page. It creates no resources and does not trigger, force, reset or recreate a dataset. Trino independently confirms those exact snapshot IDs in the table metadata and an unchanged physical column schema before and after the readback.

Set E2E_BD_REMATERIALIZATION_IDS to a comma-separated list of at least two existing dataset UUIDs, E2E_EXPECTED_FRANKCTL_VERSION to the verified installed release, and E2E_AUTH_STORAGE_PATH to private Playwright browser state. Supply the explicit E2E_API_URL, E2E_BASE_URL, E2E_TENANT_ID and E2E_FRANKCTL_PROFILE for the same deployment. The proof fails if it cannot find twenty real changed-checkpoint pairs or if definitions change during the readback. Empty histories, no-write replays and health checks alone cannot pass. Supply E2E_TRINO_URL, E2E_TRINO_USER, E2E_TRINO_CATALOG and the existing GTFS_E2E_TRINO_AUTH setting (basic or loopback-user). User-only HTTP auth is accepted only for an explicit localhost port-forward; HTTPS requires the configured Trino password.

Controlled history/current regression

The repository also provides tests/e2e-live/specs/10_sources/gtfs_realtime_semantics.spec.ts for cases a live provider cannot reliably supply. It uses fixed synthetic HTTP fixtures from the existing /api/v1/_e2e router and normal native Sources. The test changes only its owned Source's feed_url between immutable mixed, changed, empty and DIFFERENTIAL protobuf URLs, preserving its Source, feed and endpoint identities. Each poll still runs through SourceSync and writes exact bytes to Bronze; declared Pipeline patterns read those snapshots.

The test checks all ten typed projections, every nested record, history replay, empty complete-feed clearing, DIFFERENTIAL diagnostics and Source/Stream/Transform lineage. A separately owned GTFS Static Source ingests a fixed six-file ZIP; a declared SQL step joins the recorded alert's trip, route and stop references to that Source's actual Bronze tables. This recorded alert regression supplements the live matching-provider journey.

Run it after the independent capacity cohort: its four RT polls and replay execute 105 TransformRuns, and recorded alert correlation adds four more, plus 25 sandbox steps. Use the same explicit staging, published CLI, deployed SDK, Trino and private browser-state configuration as the live journey. The default fixture host is the selected API; GTFS_E2E_FIXTURE_BASE_URL may select another explicit HTTPS fixture directory. All fixture hashes are checked before any resource is created.

cd tests/e2e-live
npx playwright test specs/10_sources/gtfs_realtime_semantics.spec.ts \
  --workers=1 --reporter=line

The author helper's --realtime-only option emits the 21 native history, current and diagnostic steps without requiring Static tables. Its default mode continues to require real Static Source/table inputs for correlation. The test stores and validates those ordinary declarations through frankctl. On success it drains its exact Source, inventories every physical table, archives the Pipeline, deletes each owned Transform with cascade_lakehouse=true, and hard-resets each owned Source stream before deleting the Source. It then queries Trino information_schema.tables and records physical_tables_deleted: true. Failed proofs retain their resource definitions and inventory for diagnosis. Keep browser state, traces and raw authenticated artifacts private.