//! **Mainframe fixed-width + COBOL copybook + EBCDIC + COMP-3** collector
//! (feature `mainframe`). Decodes IBM z/OS sequential exports (DB2 UNLOAD, IDCAMS
//! REPRO, VSAM) — EBCDIC bytes, packed-decimal (COMP-3), zoned decimal, and
//! big-endian binary (COMP) — against a COBOL **copybook**, which IS the schema
//! (deterministic, no inference — the ideal inject/assert target).
//!
//! The copybook parser + field layout + Arrow schema + record slicer live in the
//! shared [`crate::fixedwidth`] module; this module adds the **IBM-037 EBCDIC
//! table** (hand-rolled — `encoding_rs` has no EBCDIC) and the numeric decode
//! primitives ([`unpack_comp3`], [`unzone`], [`binary_to_i128`]).
//!
//! Everything is hand-rolled and pure (no unvetted copybook crate), so the leaf
//! stays pure and the decode is byte-deterministic:
//! `PIC S9(3)V99 COMP-3` value `-123.45` is exactly bytes `12 34 5D`.
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow_array::RecordBatch;
use arrow_schema::Schema;
use super::fixedwidth::{ColBuilder, Field, FieldLayout, Usage, parse_copybook, slice_records};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── EBCDIC code pages ─────────────────────────────────────────────────────────
/// Supported EBCDIC code pages. IBM-037 (CP037/CCSID 037) = US/Canada EBCDIC.
/// (IBM-1047 and others are a documented seam — add a table + a variant.)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CodePage {
/// IBM-037 / CP037 — US/Canada EBCDIC.
Ibm037,
}
/// IBM-037 → Unicode code points (the full 256-byte table). Hand-rolled because
/// `encoding_rs` ships no EBCDIC; deterministic and unit-tested on the
/// letters/digits/space that fixtures exercise.
#[rustfmt::skip]
const IBM037: [u16; 256] = [
0x0000,0x0001,0x0002,0x0003,0x009C,0x0009,0x0086,0x007F,0x0097,0x008D,0x008E,0x000B,0x000C,0x000D,0x000E,0x000F,
0x0010,0x0011,0x0012,0x0013,0x009D,0x0085,0x0008,0x0087,0x0018,0x0019,0x0092,0x008F,0x001C,0x001D,0x001E,0x001F,
0x0080,0x0081,0x0082,0x0083,0x0084,0x000A,0x0017,0x001B,0x0088,0x0089,0x008A,0x008B,0x008C,0x0005,0x0006,0x0007,
0x0090,0x0091,0x0016,0x0093,0x0094,0x0095,0x0096,0x0004,0x0098,0x0099,0x009A,0x009B,0x0014,0x0015,0x009E,0x001A,
0x0020,0x00A0,0x00E2,0x00E4,0x00E0,0x00E1,0x00E3,0x00E5,0x00E7,0x00F1,0x00A2,0x002E,0x003C,0x0028,0x002B,0x007C,
0x0026,0x00E9,0x00EA,0x00EB,0x00E8,0x00ED,0x00EE,0x00EF,0x00EC,0x00DF,0x0021,0x0024,0x002A,0x0029,0x003B,0x00AC,
0x002D,0x002F,0x00C2,0x00C4,0x00C0,0x00C1,0x00C3,0x00C5,0x00C7,0x00D1,0x00A6,0x002C,0x0025,0x005F,0x003E,0x003F,
0x00F8,0x00C9,0x00CA,0x00CB,0x00C8,0x00CD,0x00CE,0x00CF,0x00CC,0x0060,0x003A,0x0023,0x0040,0x0027,0x003D,0x0022,
0x00D8,0x0061,0x0062,0x0063,0x0064,0x0065,0x0066,0x0067,0x0068,0x0069,0x00AB,0x00BB,0x00F0,0x00FD,0x00FE,0x00B1,
0x00B0,0x006A,0x006B,0x006C,0x006D,0x006E,0x006F,0x0070,0x0071,0x0072,0x00AA,0x00BA,0x00E6,0x00B8,0x00C6,0x00A4,
0x00B5,0x007E,0x0073,0x0074,0x0075,0x0076,0x0077,0x0078,0x0079,0x007A,0x00A1,0x00BF,0x00D0,0x00DD,0x00DE,0x00AE,
0x005E,0x00A3,0x00A5,0x00B7,0x00A9,0x00A7,0x00B6,0x00BC,0x00BD,0x00BE,0x005B,0x005D,0x00AF,0x00A8,0x00B4,0x00D7,
0x007B,0x0041,0x0042,0x0043,0x0044,0x0045,0x0046,0x0047,0x0048,0x0049,0x00AD,0x00F4,0x00F6,0x00F2,0x00F3,0x00F5,
0x007D,0x004A,0x004B,0x004C,0x004D,0x004E,0x004F,0x0050,0x0051,0x0052,0x00B9,0x00FB,0x00FC,0x00F9,0x00FA,0x00FF,
0x005C,0x00F7,0x0053,0x0054,0x0055,0x0056,0x0057,0x0058,0x0059,0x005A,0x00B2,0x00D4,0x00D6,0x00D2,0x00D3,0x00D5,
0x0030,0x0031,0x0032,0x0033,0x0034,0x0035,0x0036,0x0037,0x0038,0x0039,0x00B3,0x00DB,0x00DC,0x00D9,0x00DA,0x009F,
];
/// Decode an EBCDIC byte slice → a UTF-8 `String` (no trimming). One byte → one
/// Unicode scalar via the [`CodePage`] table.
pub fn ebcdic_to_utf8(bytes: &[u8], cp: CodePage) -> String {
let table = match cp {
CodePage::Ibm037 => &IBM037,
};
bytes
.iter()
.map(|&b| char::from_u32(table[b as usize] as u32).unwrap_or('\u{FFFD}'))
.collect()
}
/// Encode a `&str` → EBCDIC bytes — the exact **inverse** of [`ebcdic_to_utf8`],
/// derived from the SAME [`IBM037`] table rather than a hand-built second copy
/// (the fixtures used to carry their own partial letters/digits/space table, which
/// could drift from the decoder it was supposed to mirror). A character the code
/// page cannot represent becomes EBCDIC `?` (`0x6F`), matching how the decoder
/// renders unknown bytes.
///
/// Linear scan per character: this exists to build FIXTURES (and the offline-safe
/// mainframe sample), never to encode a production stream.
pub fn utf8_to_ebcdic(s: &str, cp: CodePage) -> Vec<u8> {
let table = match cp {
CodePage::Ibm037 => &IBM037,
};
s.chars()
.map(|c| {
let cp = c as u32;
u16::try_from(cp)
.ok()
.and_then(|want| table.iter().position(|&t| t == want))
.map(|i| i as u8)
.unwrap_or(0x6F)
})
.collect()
}
// ── numeric decode primitives ─────────────────────────────────────────────────
/// Unpack a COMP-3 (packed BCD) field → the **unscaled** `i128` mantissa. Two
/// digits per byte; the final low nibble is the sign (`C`/`F`/`A`/`E` positive,
/// `D`/`B` negative). E.g. `PIC S9(3)V99 COMP-3` `-123.45` = bytes `12 34 5D` →
/// unscaled `-12345`. Returns `None` on a bad nibble.
pub fn unpack_comp3(bytes: &[u8]) -> Option<i128> {
if bytes.is_empty() {
return None;
}
let mut val: i128 = 0;
let last = bytes.len() - 1;
for (i, &byte) in bytes.iter().enumerate() {
let hi = (byte >> 4) & 0x0F;
let lo = byte & 0x0F;
if hi > 9 {
return None;
}
val = val * 10 + hi as i128;
if i == last {
let neg = matches!(lo, 0x0D | 0x0B);
return Some(if neg { -val } else { val });
}
if lo > 9 {
return None;
}
val = val * 10 + lo as i128;
}
None
}
/// Decode a **zoned decimal** field (EBCDIC DISPLAY numeric) → unscaled `i128`.
/// One digit per byte (low nibble); the last byte's high nibble carries the sign
/// (`D`/`B` = negative). Returns `None` on a bad digit nibble.
pub fn unzone(bytes: &[u8]) -> Option<i128> {
if bytes.is_empty() {
return None;
}
let mut val: i128 = 0;
let mut neg = false;
let last = bytes.len() - 1;
for (i, &byte) in bytes.iter().enumerate() {
let hi = (byte >> 4) & 0x0F;
let lo = byte & 0x0F;
if lo > 9 {
return None;
}
if i == last {
neg = matches!(hi, 0x0D | 0x0B);
}
val = val * 10 + lo as i128;
}
Some(if neg { -val } else { val })
}
/// Decode a big-endian two's-complement COMP/BINARY field → `i128`.
pub fn binary_to_i128(bytes: &[u8], signed: bool) -> i128 {
let mut val: i128 = 0;
for &b in bytes {
val = (val << 8) | b as i128;
}
if signed && !bytes.is_empty() {
let bits = (bytes.len() * 8) as u32;
let sign_bit = 1i128 << (bits - 1);
if val & sign_bit != 0 {
val -= 1i128 << bits; // sign-extend
}
}
val
}
// ── record decode ─────────────────────────────────────────────────────────────
/// Decode one elementary field's byte slice into its column builder (EBCDIC).
fn decode_field(f: &Field, slice: &[u8], cp: CodePage, col: &mut ColBuilder) {
if f.pic.is_text {
let s = ebcdic_to_utf8(slice, cp);
col.push_text(Some(s.trim_end().to_string()));
return;
}
let unscaled = match f.usage {
Usage::Comp3 => unpack_comp3(slice),
Usage::Display => unzone(slice),
Usage::Comp => Some(binary_to_i128(slice, f.pic.signed)),
// COMP-1/COMP-2 (float) are a seam: decode as null rather than mis-cast.
Usage::Comp1 | Usage::Comp2 => None,
};
col.push_int(unscaled);
}
/// Decode an **EBCDIC** fixed-width blob against a copybook layout → Arrow. This
/// is the core inject/assert entry point: bytes + copybook → typed `RecordBatch`
/// (COMP-3 → correct `Decimal128`).
pub fn decode_ebcdic_records(
layout: &FieldLayout,
bytes: &[u8],
cp: CodePage,
) -> Result<RecordBatch> {
let records = slice_records(bytes, layout.lrecl);
let mut builders: Vec<ColBuilder> = layout.fields.iter().map(ColBuilder::for_field).collect();
for rec in &records {
for (f, b) in layout.fields.iter().zip(builders.iter_mut()) {
decode_field(f, &rec[f.offset..f.offset + f.len], cp, b);
}
}
let columns = builders
.into_iter()
.map(ColBuilder::finish)
.collect::<Vec<_>>();
Ok(RecordBatch::try_new(Arc::new(layout.schema()), columns)?)
}
// ── collector ─────────────────────────────────────────────────────────────────
/// The EBCDIC + COBOL-copybook mainframe collector. `cfg.params.copybook` is the
/// COBOL layout (the schema); `cfg.base_url` is the EBCDIC data file. `gdpr` is
/// [`GdprClass::Personal`] by default — mainframe master files are PII-heavy.
pub struct MainframeCollector {
codepage: CodePage,
}
impl Default for MainframeCollector {
fn default() -> Self {
Self {
codepage: CodePage::Ibm037,
}
}
}
fn layout_from_cfg(cfg: &CollectorCfg) -> Result<FieldLayout> {
let text = cfg
.params
.get("copybook")
.and_then(|v| v.as_str())
.ok_or_else(|| {
anyhow!("mainframe collector needs `params.copybook` (COBOL copybook text)")
})?;
parse_copybook(text)
}
#[async_trait::async_trait]
impl Collector for MainframeCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "mainframe-copybook".into(),
direction: Direction::Source,
source_kind: SourceKind::FlatFile,
path: CollectorPath::RustNative,
formats: vec!["ebcdic-fixed".into(), "cobol-copybook".into()],
license: "per-source".into(),
gdpr: GdprClass::Personal,
auth: AuthKind::None,
join_keys: vec![],
description: "IBM z/OS EBCDIC fixed-width records decoded against a COBOL \
copybook (PIC X/9/S9V, COMP-3 packed, zoned, COMP binary) → Arrow \
(COMP-3 → Decimal128)."
.into(),
}
}
/// The shared [`crate::fixedwidth::supplied_file_card`] — same absent authority,
/// same copybook-is-the-schema rules, same LRECL traps, written once next door —
/// plus the three facts that are EBCDIC's alone: one code page, one transfer mode
/// that must not be got wrong, and a GDPR class that is `Personal` and stays there.
///
/// `docs_url` is this module's own source because the decode contract an operator
/// needs is stated there in bytes (`PIC S9(3)V99 COMP-3` value `-123.45` is exactly
/// `12 34 5D`), and no vendor page states korp's supported subset.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
Some(
super::fixedwidth::supplied_file_card(
"https://git.gunnar.rs/nordisk/korp-collectors",
)
.with_quirk(
"TRANSFER THE DATASET IN BINARY. An EBCDIC export pulled off z/OS in FTP \
ASCII mode has already been transcoded — and once packed COMP-3 fields \
have been run through a character translation the original bytes are \
gone. This collector cannot detect that; it will decode the damage.",
)
.with_quirk(
"IBM-037 (CP037, US/Canada EBCDIC) is the ONLY code page implemented — \
the `CodePage` enum has a single variant and the registered collector is \
built from `Default`, so there is nothing to configure. A dataset in \
another page (IBM-1047, or the Nordic 278 with its own åäö positions) \
decodes to wrong letters SILENTLY: the bytes are all valid, they just \
mean something else. Check the åäö before trusting a load.",
)
.with_quirk(
"GDPR class is `Personal` and hardcoded in `meta()`, not derived — a z/OS \
sequential export (DB2 UNLOAD, IDCAMS REPRO, VSAM) is assumed to carry \
personal data. Nothing in configuration relaxes it.",
)
.with_quirk(
"COMP-3 lands as Arrow `Decimal128` with the copybook's own scale, not as \
a float: `PIC S9(3)V99 COMP-3` holding `-123.45` is exactly the bytes \
`12 34 5D`, and the sign is the LAST nibble (`D`/`B` negative, \
`C`/`F`/`A`/`E` positive). A bad nibble becomes NULL rather than an \
error, so a column of nulls means the field's `USAGE` in your copybook \
does not match the bytes.",
),
)
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(SourceEndpoint::FlatFile, "", "mainframe_import")
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
Ok(layout_from_cfg(cfg)?.schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let layout = layout_from_cfg(cfg)?;
let bytes = crate::http::read_source(cfg, &cfg.base_url).await?;
let rb = decode_ebcdic_records(&layout, &bytes, self.codepage)?;
let rows = rb.num_rows() as u64;
if rows == 0 {
return Ok(CollectStats {
reason: Some(format!(
"mainframe `{}` sliced to 0 records (LRECL={}, {} bytes)",
cfg.base_url,
layout.lrecl,
bytes.len()
)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}