FAILURE MAP
← Case archive

FA-47996 / Delimited text / Open access

Imported rows are labeled with the wrong schema revision · case 01

A structured table violates the declared record or column contract.

Verified by executionVariant 1 · 8 checks per implementationDownload source bundle ↓JSON ↗

ROOT CAUSE

Result metadata uses schema installation order instead of row-declared identity.

VERIFIED REPAIR

Preserve the named invariant at the faulty decision: rows.append([version, dict(zip(header,values))])

Unsuccessful approach: Selecting the oldest installation also mislabels out-of-order references.

Case contract

Read tagged pipe rows. S|version|headers installs a schema; D|version|values binds a data row to the matching installed version. Revisions are immutable, may be referenced out of installation order, and headers must be unique. Reject unknown version, duplicate revision, or incorrect row width. Return version plus named rows.

Why this case matters

Delimited interchange needs explicit framing, schema and field semantics at ingestion and emission boundaries.

1 / The failure

Exit 1
"""Failure Map reference implementation. Python standard library only."""
import json
def _vary(value):
    if value == '@END': return 3 + 4*N
    if isinstance(value, str): return value.replace('@', 'cell' * N)
    if isinstance(value, list): return [_vary(x) for x in value]
    if isinstance(value, dict): return {_vary(k): _vary(v) for k,v in value.items()}
    return value
N = 1
observations = []
def solve(data):
    schemas, rows = {}, []
    for line in data:
        cells = line.split('|')
        if len(cells) < 3: return None
        tag, version, values = cells[0], cells[1], cells[2:]
        if tag == 'S':
            if version in schemas or len(set(values)) != len(values): return None
            schemas[version] = values
        elif tag == 'D':
            if version not in schemas: return None
            header = schemas[version]
            if len(values) != len(header): return None
            rows.append([list(schemas)[-1], dict(zip(header,values))])
        else: return None
    return rows
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('versions', solve(_vary(['S|unused|x', 'S|1|a|b', 'D|1|@|z', 'S|2|b|a', 'D|2|y|x', 'D|1|m|n'])), _vary([['1', {'a': '@', 'b': 'z'}], ['2', {'b': 'y', 'a': 'x'}], ['1', {'a': 'm', 'b': 'n'}]]))
check('unknown', solve(_vary(['D|1|@'])), _vary(None))
check('immutable', solve(_vary(['S|1|a', 'S|1|b'])), _vary(None))
check('duplicate header', solve(_vary(['S|1|a|a'])), _vary(None))
check('wrong width', solve(_vary(['S|1|a|b', 'D|1|@'])), _vary(None))
check('normal', solve(_vary(['S|@|a', 'D|@|z'])), _vary([['@', {'a': 'z'}]]))
check('bad tag', solve(_vary(['X|1|a'])), _vary(None))
check('empty', solve(_vary([])), _vary([]))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
Boundary fixtureActualExpectedOutcome
versions[['1', {'a': 'cell', 'b': 'z'}], ['2', {'a': 'x', 'b': 'y'}], ['2', {'a': 'm', 'b': 'n'}]][['1', {'a': 'cell', 'b': 'z'}], ['2', {'a': 'x', 'b': 'y'}], ['1', {'a': 'm', 'b': 'n'}]]Failed
unknownNoneNonePassed
immutableNoneNonePassed
duplicate headerNoneNonePassed
wrong widthNoneNonePassed
normal[['cell', {'a': 'z'}]][['cell', {'a': 'z'}]]Passed
bad tagNoneNonePassed
empty[][]Passed

SHA-256 / 38d84c111fda592232032a6625b492a36d5518b306aef1d1027901efe6ab7b7e

2 / The unsuccessful fix

Exit 1
"""Failure Map reference implementation. Python standard library only."""
import json
def _vary(value):
    if value == '@END': return 3 + 4*N
    if isinstance(value, str): return value.replace('@', 'cell' * N)
    if isinstance(value, list): return [_vary(x) for x in value]
    if isinstance(value, dict): return {_vary(k): _vary(v) for k,v in value.items()}
    return value
N = 1
observations = []
def solve(data):
    schemas, rows = {}, []
    for line in data:
        cells = line.split('|')
        if len(cells) < 3: return None
        tag, version, values = cells[0], cells[1], cells[2:]
        if tag == 'S':
            if version in schemas or len(set(values)) != len(values): return None
            schemas[version] = values
        elif tag == 'D':
            if version not in schemas: return None
            header = schemas[version]
            if len(values) != len(header): return None
            rows.append([list(schemas)[0], dict(zip(header,values))])
        else: return None
    return rows
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('versions', solve(_vary(['S|unused|x', 'S|1|a|b', 'D|1|@|z', 'S|2|b|a', 'D|2|y|x', 'D|1|m|n'])), _vary([['1', {'a': '@', 'b': 'z'}], ['2', {'b': 'y', 'a': 'x'}], ['1', {'a': 'm', 'b': 'n'}]]))
check('unknown', solve(_vary(['D|1|@'])), _vary(None))
check('immutable', solve(_vary(['S|1|a', 'S|1|b'])), _vary(None))
check('duplicate header', solve(_vary(['S|1|a|a'])), _vary(None))
check('wrong width', solve(_vary(['S|1|a|b', 'D|1|@'])), _vary(None))
check('normal', solve(_vary(['S|@|a', 'D|@|z'])), _vary([['@', {'a': 'z'}]]))
check('bad tag', solve(_vary(['X|1|a'])), _vary(None))
check('empty', solve(_vary([])), _vary([]))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
Boundary fixtureActualExpectedOutcome
versions[['unused', {'a': 'cell', 'b': 'z'}], ['unused', {'a': 'x', 'b': 'y'}], ['unused', {'a': 'm', 'b': 'n'}]][['1', {'a': 'cell', 'b': 'z'}], ['2', {'a': 'x', 'b': 'y'}], ['1', {'a': 'm', 'b': 'n'}]]Failed
unknownNoneNonePassed
immutableNoneNonePassed
duplicate headerNoneNonePassed
wrong widthNoneNonePassed
normal[['cell', {'a': 'z'}]][['cell', {'a': 'z'}]]Passed
bad tagNoneNonePassed
empty[][]Passed

SHA-256 / 63d9847ba8900857dca42c3b3c9a7d5a375d2a135f9ff26dcd6bc06fa2594c5a

3 / The verified repair

Exit 0
"""Failure Map reference implementation. Python standard library only."""
import json
def _vary(value):
    if value == '@END': return 3 + 4*N
    if isinstance(value, str): return value.replace('@', 'cell' * N)
    if isinstance(value, list): return [_vary(x) for x in value]
    if isinstance(value, dict): return {_vary(k): _vary(v) for k,v in value.items()}
    return value
N = 1
observations = []
def solve(data):
    schemas, rows = {}, []
    for line in data:
        cells = line.split('|')
        if len(cells) < 3: return None
        tag, version, values = cells[0], cells[1], cells[2:]
        if tag == 'S':
            if version in schemas or len(set(values)) != len(values): return None
            schemas[version] = values
        elif tag == 'D':
            if version not in schemas: return None
            header = schemas[version]
            if len(values) != len(header): return None
            rows.append([version, dict(zip(header,values))])
        else: return None
    return rows
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('versions', solve(_vary(['S|unused|x', 'S|1|a|b', 'D|1|@|z', 'S|2|b|a', 'D|2|y|x', 'D|1|m|n'])), _vary([['1', {'a': '@', 'b': 'z'}], ['2', {'b': 'y', 'a': 'x'}], ['1', {'a': 'm', 'b': 'n'}]]))
check('unknown', solve(_vary(['D|1|@'])), _vary(None))
check('immutable', solve(_vary(['S|1|a', 'S|1|b'])), _vary(None))
check('duplicate header', solve(_vary(['S|1|a|a'])), _vary(None))
check('wrong width', solve(_vary(['S|1|a|b', 'D|1|@'])), _vary(None))
check('normal', solve(_vary(['S|@|a', 'D|@|z'])), _vary([['@', {'a': 'z'}]]))
check('bad tag', solve(_vary(['X|1|a'])), _vary(None))
check('empty', solve(_vary([])), _vary([]))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
Boundary fixtureActualExpectedOutcome
versions[['1', {'a': 'cell', 'b': 'z'}], ['2', {'a': 'x', 'b': 'y'}], ['1', {'a': 'm', 'b': 'n'}]][['1', {'a': 'cell', 'b': 'z'}], ['2', {'a': 'x', 'b': 'y'}], ['1', {'a': 'm', 'b': 'n'}]]Passed
unknownNoneNonePassed
immutableNoneNonePassed
duplicate headerNoneNonePassed
wrong widthNoneNonePassed
normal[['cell', {'a': 'z'}]][['cell', {'a': 'z'}]]Passed
bad tagNoneNonePassed
empty[][]Passed

SHA-256 / e7078e9ff823c24eec1191595019885569342ff434b22e8e34604bf818424098

Verification & scope

Deterministic bounded in-memory model. No claim of complete CSV or external format conformance. This reproducer isolates one failure mechanism. Results cover the supplied fixtures. Variants within a family share a test contract and should remain grouped when constructing evaluation splits. Related mechanisms with a shared evaluation_group must also remain together; these controlled models are not independent production incidents.

Observations recorded using Python 3.12.14 at 2026-09-29T14:44:46.795173+00:00.

Case digest / a927e0b791f8bfd4670cff7c6b096c8aaaa99efb13d2bdf66b690716e51da1a8