Table-driven stream discovery

Let a live table choose which stream members to record, while each sealed segment keeps the roster and context that selected it.

Use a table_members series when a changing catalog, schedule or watchlist determines what a stream subscription should follow. The upstream table must already contain the venue’s actual member IDs. A team name, event ID or display URL does not automatically identify an exchange market. Resolve that relationship upstream and leave uncertain matches out of the catalog.

This is a configurable pattern, not a built-in sports, weather or market integration.

This recipe starts after market identity has been resolved. It does not turn team names, schedule IDs or event labels into Kalshi or Polymarket member IDs. For a sports build, keep the schedule and the venue matching evidence as separate inputs, then publish a catalog whose member column contains the exact IDs accepted by the connector. Register and run that catalog before binding the stream series, and verify the selected member count and contexts in the subscription status before treating the feed as covered.

What this pattern is good for

Use case	Input table contains	Stream table keeps
Sports markets	exchange market IDs, game ID and team labels	order-book frames with the game context active when each segment was recorded
Weather contracts	contract IDs, city, date and threshold	quote or trade frames with those labels
A market watchlist	the members an analyst selected and optional tags	a bounded feed for the current selection, with historical tags preserved
Economic events	release or decision contract IDs, event name and date	market activity attributed to the event catalog version
Research cohorts	instrument IDs and cohort labels	observations partitioned by the cohort that selected them

The table supplies selection and metadata. It does not fetch the stream, perform entity matching, run SQL inside the recorder or create a URL for each row.

The data flow

live input table → Studio resolves one live version → bounded projection
  → sealed roster/context snapshot → stream subscription → output recipe

The input table must be live and in the same workspace. A pinned input follows its pin. A successful empty refresh is different from a failed read: empty input removes the roster and leaves the subscription waiting; a failed read retains the last verified roster only until its freshness deadline. When that deadline expires, collection stops and reports the gap.

1. Prepare the input table

Create and publish a table with a UTF-8 member column and optional string, integer or boolean context columns. Every selected member and context value must be present. If several rows project the same member, their projected context must agree; conflicting context refuses the whole input. Input row order does not matter.

For an NBA-style catalog, these are deliberately fictional test identifiers. This table is a shape example only; it cannot be registered until the IDs, source and rights evidence are replaced with real values:

market_id	game_id	away_team	home_team
NBA-FIXTURE-001-HOME	game-001	AWAY-A	HOME-A
NBA-FIXTURE-002-HOME	game-002	AWAY-B	HOME-B

Replace them with real IDs from the venue before recording. A single game can have several market or outcome rows, so market_id is the subscription member while game_id is context.

For a schedule-first workflow, use these stages:

1. Build and verify the schedule table first. Keep the provider game ID, scheduled start, teams, status and source provenance.
2. Resolve schedule games to venue event and market IDs in a separate matching step. Preserve unmatched and ambiguous candidates for review; do not invent a member ID from a team name or slug.
3. Build the bounded venue catalog with one row per connector member and the game context that selected it. Check member uniqueness and non-null required context.
4. Register the table_members series against that catalog, then bind and sample the recorded stream recipe. Confirm member count, discovery freshness, snapshots, updates and composite event uniqueness before claiming continuous coverage.

The current recipe surface supports these as separate tables and joins. It does not provide a single recipe field that performs arbitrary per-row venue requests or schedule-to-market matching.

2. Register the series

The catalog is an ordinary live table, so build it with a normal recipe first. This compact input recipe shows the important contract: a stable market_id plus typed context and a refreshable source. Replace the URL, digest and SQL with the source that performs your entity matching.

catalog-recipe.json

{
  "dataset": {
    "name": "Market catalog",
    "description": "Venue member IDs matched to events and labels by an upstream process."
  },
  "question": { "text": "Which venue members should the subscription follow?" },
  "table": {
    "name": "market_catalog",
    "description": "One row per matched venue member and its context.",
    "grain": ["market_id"],
    "columns": [
      { "name": "market_id", "type": "string", "nullable": false, "description": "The venue's actual subscription member ID." },
      { "name": "game_id", "type": "string", "nullable": false, "description": "The upstream event identifier." },
      { "name": "away_team", "type": "string", "nullable": false, "description": "The away team label." },
      { "name": "home_team", "type": "string", "nullable": false, "description": "The home team label." }
    ]
  },
  "sources": [{
    "name": "matched_catalog",
    "source_class": "user_url",
    "data_classification": "public",
    "locator": { "kind": "https_url", "display_locator": "https://catalog.example.invalid/matched-markets.json" },
    "rights_claim": { "claimed_basis": "unknown", "claim_evidence_digest": "sha256:REPLACE_WITH_64_LOWERCASE_HEX_CHARS" },
    "connector": { "adapter_id": "public.https", "credential_mode": "none" },
    "limits": { "max_source_bytes": 1048576 }
  }],
  "transform": {
    "engine": "duckdb_sql",
    "steps": [{
      "step_id": "market_catalog",
      "sql": "select market_id, game_id, away_team, home_team from matched_catalog",
      "description": "Return the member ID and every projected context field as the declared types."
    }]
  },
  "checks": [
    { "check_id": "member_unique", "kind": "key_uniqueness", "enforcement": "required", "columns": ["market_id"] },
    { "check_id": "catalog_nonempty", "kind": "row_expectation", "enforcement": "required", "min_rows": 1 }
  ],
  "units": [],
  "timezone": "UTC"
}

Register and run this input recipe through Run and inspect, then use its resulting live table_id in the series document below. An input recipe that contains only event names is insufficient: the member column must contain the IDs accepted by the connector.

Save this as table-series.json. Replace the connector digest, input table_id, member column and metadata fields with values from your own tables and connector document. The UUID below is a shape placeholder, not a usable table.

table-series.json

{
  "schema_version": "harness-stream-series.v7",
  "series": {
    "schema_version": "harness-stream-series.v7",
    "series_id": "market-catalog-series",
    "connector_id": "example.market-feed",
    "member_stream_id_prefix": "catalog-stream",
    "selection_rule": "table_members",
    "table_discovery": {
      "table_id": "11111111-1111-4111-8111-111111111111",
      "member_id_column": "market_id",
      "interval_seconds": 60,
      "stale_after_seconds": 3600,
      "max_members": 32,
      "max_rows": 128,
      "max_bytes": 1048576,
      "max_parts": 8,
      "member_metadata": {
        "event_member_selectors": [{ "pointer": "/market_id" }],
        "fields": [
              { "name": "away_team", "column": "away_team", "kind": "string" },
              { "name": "game_id", "column": "game_id", "kind": "string" },
              { "name": "home_team", "column": "home_team", "kind": "string" }
        ]
      }
    },
    "pre_subscribe_seconds": 0,
    "settlement_seconds": 0,
    "settlement_grace_seconds": 0,
    "max_concurrent_members": 32
  }
}

event_member_selectors identifies the member in the connector’s normalized event payload. The selector must match the connector document; /market_id is only an example. For multi-member events, use the connector’s supported array selector syntax. The metadata kind is string, integer or boolean and must agree with the input table’s representation.

Register it against the existing connector document:

mr-data stream series register --file table-series.json \
  --digest CONNECTOR_DOCUMENT_DIGEST --json
mr-data stream subscribe --registry REGISTRY_ID \
  --document-digest CONNECTOR_DOCUMENT_DIGEST \
  --series market-catalog-series --json
mr-data stream subscription show SUBSCRIPTION_ID --json

The series registration returns the registry and series identifiers to use in the subscription command. In the JSON receipt, use stream_connector_registry_id as REGISTRY_ID; keep the registered series ID as SERIES_ID if the service returns a generated one. Keep the venue credential setup on the connector; source-table credentials are not placed in this document.

3. Bind the output recipe

An output recipe is an ordinary recorded-stream recipe. This complete shape keeps one row per recorded event and carries the member needed to match the sealed context snapshot. Replace every REPLACE_* value before registering it; the placeholders make this a template rather than a claim that the example endpoint or digest exists.

output-recipe.json

{
  "dataset": {
    "name": "Recorded market frames with catalog context",
    "description": "One row per stream event, with its subscribed member and the segment snapshot that supplies context."
  },
  "question": { "text": "What did each selected market publish, and which catalog context selected it?" },
  "table": {
    "name": "market_frames",
    "description": "One row per recorded event. Match member to the segment's member_contexts snapshot when reading context.",
    "grain": ["source_segment_id", "event_id"],
    "columns": [
      { "name": "source_segment_id", "type": "string", "nullable": false, "description": "The sealed recorder segment; part of the event identity." },
      { "name": "event_id", "type": "string", "nullable": false, "description": "Connection-local recorder event ID; unique only with source_segment_id." },
      { "name": "member", "type": "string", "nullable": false, "description": "The venue member attributed by the connector." },
      { "name": "received_at", "type": "timestamp", "nullable": false, "description": "UTC time the recorder received the event." },
      { "name": "message", "type": "string", "nullable": false, "description": "Canonical JSON provider payload." }
    ]
  },
  "sources": [{
    "name": "market_stream",
    "description": "Recorded frames from the venue subscription selected by the market catalog.",
    "source_class": "stream",
    "data_classification": "public",
    "locator": { "kind": "stream", "display_locator": "wss://REPLACE_VENUE_HOST/feed" },
    "rights_claim": {
      "claimed_basis": "terms_of_service_asserted",
      "claim_evidence_digest": "sha256:REPLACE_WITH_64_LOWERCASE_HEX_CHARS"
    },
    "connector": {
      "adapter_id": "stream.wss@1.0.0",
      "credential_mode": "none",
      "stream_connector_document_digest": "REPLACE_CONNECTOR_DOCUMENT_DIGEST",
      "parameters": [{ "name": "stream.series", "value": "REPLACE_SERIES_ID" }]
    },
    "stream_relation_version": "v2",
    "limits": { "max_source_bytes": 268435456, "max_rows": 5000000 }
  }],
  "transform": {
    "engine": "duckdb_sql",
    "steps": [{
      "step_id": "market_frames",
      "sql": "select source_segment_id, event_id, member, cast(received_at as timestamptz) as received_at, message from market_stream",
      "description": "Keep the recorder's composite event key, attributed member, arrival time and canonical payload."
    }]
  },
  "checks": [
    { "check_id": "event_key_unique", "kind": "key_uniqueness", "enforcement": "required", "columns": ["source_segment_id", "event_id"] },
    { "check_id": "member_present", "kind": "null_ceiling", "enforcement": "required", "columns": ["member"], "max_null_ppm": 0 },
    { "check_id": "frames_present", "kind": "row_expectation", "enforcement": "required", "min_rows": 1 }
  ],
  "units": [],
  "timezone": "UTC"
}

Use stream.wss@2.0.0 and the connector’s shaped credential when the venue is keyed. The source declares stream.series or stream.members, never both. Register, sample, inspect and run this recipe through the normal build loop. The receipt includes the registry ID bound to the series; preserve that value when debugging a subscription.

The output table’s source_segment_id and event_id are the composite event grain. The member value is the join key for the member_contexts snapshot sealed with that segment. Read the snapshot from the run’s provenance/receipt when presenting game_id, team labels or other projected fields; do not join an old event to the current input table, whose roster may have changed.

What the output relation contains

The recorder chooses the stream relation shape. A connector with a member field exposes a ninth member column after event_time; otherwise the relation has the original eight columns. For table-driven discovery, use the member-bearing shape so the event can be joined to its projected context.

The output’s stream relation is version 2 for incremental continuation. Treat source_segment_id and event_id as the composite event identity: event_id is connection-local and is not globally unique. The sealed segment also carries the member_contexts snapshot used to record it. Join an event’s member to that snapshot when a frame contains more than one member. Never look up old context from the current input table during replay.

select source_segment_id, event_id, member, received_at, message
from market_frames
where message_type = 'delta'

Changes, limits and recovery

• New members and changed context take effect at a sealed segment boundary after the next successful discovery. Earlier frames retain their original snapshot.
• A different input version with the same sorted member/context set does not force a reconnect.
• Limits apply before deduplication and include physical parts, bytes, rows and selected members. The retained canonical snapshot has its own 16 MiB ceiling, including metadata and provenance. A wide projection can therefore admit fewer members than max_members.
• Malformed, oversized or unavailable input is refused as a whole. A successful empty result leaves an idle subscription; it is not an input failure.
• A stale or failed dependency stops discovery after the freshness deadline. Updating and successfully verifying the input materialization lets the subscription recover.
• Self-dependencies and transitive dependency cycles are refused. The input table cannot depend on the output recipe that consumes it.
• A subscription’s healthy lease does not prove discovery found members or that the venue delivered frames. Inspect member count, discovery timestamps and gap fields in subscription status.

The sealed snapshot is retained with each segment, so roster and context replay remains reproducible after the input table version expires. It cannot recreate messages from a period when no recorder was connected.

Related pages

• Live streams — recorder lifecycle, relation columns and limits.
• Source kinds and connectors — stream.series, credentials and connector rules.
• WebSocket stream — a complete fixed-members recipe.
• Stream to bars — a member-bearing stream transform.
• Recipe document — the full recipe grammar.