Build a mapping step

A mapping step merges equivalent keys from multiple upstream runs, writes a canonical registry, and writes a source-run/local-id crosswalk. The merge algorithm belongs to framework mapping.

1. Declare upstream runs

params_defaults:
  mapping.upstream_runs: ""
  mapping.run_id: ""

Expose mapping.upstream_runs as a runs picker in ui.streamlit when launched interactively.

2. Declare datasets and hash merge

drtml_version: 3
step_id: merge_tokens
pattern: mapping

mapping:
  entry: drtoller.framework.processing.mapping_mode.run:run_mapping
  inputs:
    source: token_registry_run
  outputs:
    registry: token_registry_canonical
    mapping: token_mapping

storage:
  bindings:
    artifacts: {backend: s3, root: pipeline-artifacts, profile: minio_app}

datasets:
  token_registry_run:
    role: input
    binding: artifacts
    run_id_param: mapping.upstream_runs
    columns:
      token_id_run: string
      language: string
      token: string
    merge:
      method: hash
      key_columns:
        language: string
        token: string
      source_id_column: token_id_run
      hash_column: token_id

  token_registry_canonical:
    role: output
    binding: artifacts
    columns:
      token_id: string
      language: string
      token: string

  token_mapping:
    role: output
    binding: artifacts
    columns:
      source_run_id: string
      token_id_run: string
      token_id: string

Current merge method is hash. Exactly one input may declare merge. The mapping compiler expects exactly one output dataset whose id contains registry and one containing mapping, then validates canonical id columns.

For hash-local contracts where source_id_column == hash_column, the hash column is selected as canonical id explicitly.

3. Add a thin hook only if needed

from drtoller.framework.processing.mapping_mode.run import run_mapping

def run(ctx, *, manifest_path):
    return run_mapping(ctx, manifest_path=manifest_path)

No step-specific merge implementation is required.

4. Consume a mapping downstream

params_defaults:
  mapping.token_map_run: ""

mapping_read:
  tokens:
    dataset: token_mapping
    run_id_param: mapping.token_map_run
    local_id_column: token_id_run
    canonical_id_column: token_id
    source_run_column: source_run_id

processing:
  mapping_read_bindings:
    token_index: tokens

Framework prefetches an IdIndex and injects it into processor context. Input columns that declare remap: tokens require the matching alias.

Domain code remaps by id only using framework mapping helpers. It does not reopen crosswalk datasets or reimplement merge.

5. Test

• Manifest compiles and identifies merge input, registry output and mapping output.
• Same key across runs receives one canonical id.
• Crosswalk preserves (source_run_id, local_id) → canonical_id.
• Rerun is deterministic.
• Downstream mapping_read resolves the correct run and injects the index.
• UI/metrics include live params and CPU/RAM panels.