//! **Fixed-width flat files + the COBOL-copybook layout parser** (feature
//! `mainframe`). This module carries the parts shared by the EBCDIC mainframe
//! collector ([`crate::mainframe`]) and its ASCII cousin: the COBOL **copybook
//! PICTURE/USAGE parser**, the derived **field layout = Arrow schema** (the
//! copybook IS the schema — deterministic, no inference), the fixed-`LRECL`
//! record slicer, and the ASCII decode path.
//!
//! The mainframe module adds the EBCDIC table + packed/zoned/binary decode on top
//! of the exact same [`FieldLayout`]; #3 in the catalog (ASCII fixed-width) falls
//! out here for near-free via [`FixedWidthCollector`].
use anyhow::{Result, anyhow};
use arrow_array::builder::{Decimal128Builder, Int32Builder, Int64Builder, StringBuilder};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::{DataType, Field as ArrowField, Schema};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── PICTURE + USAGE ───────────────────────────────────────────────────────────
/// The decoded facts of a COBOL `PIC` clause.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PicInfo {
/// Alphanumeric (`X`/`A`) — a text field of `x_len` chars.
pub is_text: bool,
/// `S` present — signed numeric.
pub signed: bool,
/// Integer digit count (before an implied `V`).
pub int_digits: usize,
/// Fractional digit count (after `V`) — the Decimal **scale**.
pub frac_digits: usize,
/// Char length for a text (`X`) field.
pub x_len: usize,
}
impl PicInfo {
/// Total significant digits (numeric precision `p = int + frac`).
pub fn digits(&self) -> usize {
self.int_digits + self.frac_digits
}
}
/// COBOL `USAGE`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Usage {
/// `DISPLAY` — text, or **zoned** decimal for numerics (1 byte/digit).
Display,
/// `COMP-3` / `PACKED-DECIMAL` — BCD, 2 digits/byte, sign nibble last.
Comp3,
/// `COMP` / `COMP-4` / `BINARY` — big-endian two's-complement.
Comp,
/// `COMP-1` — 4-byte float.
Comp1,
/// `COMP-2` — 8-byte float.
Comp2,
}
/// Parse a COBOL picture string (`X(5)`, `S9(5)V99`, `9(4)`, `999`, `S9(7)V9(2)`)
/// into a [`PicInfo`]. `(n)` repeats the preceding symbol; `V` is the implied
/// decimal point; `S` marks a sign; `P` (scaling) is tolerated but ignored (seam).
pub fn parse_pic(pic: &str) -> PicInfo {
let chars: Vec<char> = pic.to_ascii_uppercase().chars().collect();
let (mut signed, mut seen_v) = (false, false);
let (mut int_digits, mut frac_digits, mut x_len) = (0usize, 0usize, 0usize);
let mut last: Option<char> = None; // '9' or 'X'
let mut i = 0;
while i < chars.len() {
match chars[i] {
'S' => signed = true,
'V' => seen_v = true,
'9' => {
if seen_v {
frac_digits += 1
} else {
int_digits += 1
}
last = Some('9');
}
'X' | 'A' => {
x_len += 1;
last = Some('X');
}
'P' => {} // scaling position — ignored (documented seam)
'(' => {
let mut j = i + 1;
let mut num = String::new();
while j < chars.len() && chars[j] != ')' {
num.push(chars[j]);
j += 1;
}
let n: usize = num.trim().parse().unwrap_or(1);
let extra = n.saturating_sub(1); // the symbol was already counted once
match last {
Some('9') => {
if seen_v {
frac_digits += extra
} else {
int_digits += extra
}
}
Some('X') => x_len += extra,
_ => {}
}
i = j; // loop `+= 1` lands past ')'
}
_ => {}
}
i += 1;
}
PicInfo {
is_text: x_len > 0,
signed,
int_digits,
frac_digits,
x_len,
}
}
// ── field layout (copybook → schema) ──────────────────────────────────────────
/// One elementary field, positioned in the fixed record.
#[derive(Debug, Clone)]
pub struct Field {
pub name: String,
/// Byte offset within the record.
pub offset: usize,
/// Byte length of this field.
pub len: usize,
pub usage: Usage,
pub pic: PicInfo,
/// The Arrow type this field decodes to (spec-driven, no inference).
pub arrow_type: DataType,
}
/// The full record layout parsed from a copybook = the Arrow schema + `LRECL`.
#[derive(Debug, Clone)]
pub struct FieldLayout {
pub fields: Vec<Field>,
/// Fixed record length (`RECFM=FB`) in bytes.
pub lrecl: usize,
}
impl FieldLayout {
/// The Arrow schema — one nullable field per elementary field. (Fixed fields
/// are structurally present, but we keep them nullable to tolerate
/// uninitialized / space-filled records rather than panic.)
pub fn schema(&self) -> Schema {
Schema::new(
self.fields
.iter()
.map(|f| ArrowField::new(&f.name, f.arrow_type.clone(), true))
.collect::<Vec<_>>(),
)
}
}
/// Arrow type for a `(pic, usage)` pair — the catalog §2 mapping.
pub fn arrow_type_of(pic: &PicInfo, usage: Usage) -> DataType {
if pic.is_text {
return DataType::Utf8;
}
if pic.frac_digits > 0 {
// implied decimal → Decimal128(p+s, s)
return DataType::Decimal128(pic.digits() as u8, pic.frac_digits as i8);
}
match usage {
Usage::Comp1 => DataType::Float32,
Usage::Comp2 => DataType::Float64,
Usage::Comp => {
if pic.int_digits <= 4 {
DataType::Int32
} else if pic.int_digits <= 18 {
DataType::Int64
} else {
DataType::Decimal128(pic.int_digits as u8, 0)
}
}
_ => {
if pic.int_digits <= 18 {
DataType::Int64
} else {
DataType::Decimal128(pic.int_digits as u8, 0)
}
}
}
}
/// Byte length of a `(pic, usage)` field.
pub fn len_of(pic: &PicInfo, usage: Usage) -> usize {
if pic.is_text {
return pic.x_len;
}
let digits = pic.digits();
match usage {
Usage::Display => digits, // zoned: 1 byte/digit
Usage::Comp3 => digits / 2 + 1, // packed: 2 digits/byte + sign nibble
Usage::Comp1 => 4,
Usage::Comp2 => 8,
Usage::Comp => {
if digits <= 4 {
2
} else if digits <= 9 {
4
} else {
8
} // big-endian binary
}
}
}
/// A raw copybook entry (one statement) before offsets are computed.
struct RawEntry {
/// Level number (01/05/10…). Consumed by the REDEFINES resolver: the subject
/// of a `REDEFINES` shares the redefining item's level, so `level`
/// disambiguates which prior entry an overlay targets.
level: u8,
name: String,
pic: Option<PicInfo>,
usage: Usage,
occurs: usize,
/// `Some(target)` when this entry `REDEFINES target` — it overlays `target`'s
/// bytes (placed at `target`'s offset; the record cursor does NOT advance for
/// it). `None` for a normal field.
redefines: Option<String>,
}
/// Parse copybook `text` → a [`FieldLayout`] (elementary fields with computed
/// offsets + `LRECL`). Handles level numbers, PIC X/9/S9/V, USAGE
/// DISPLAY/COMP-3/COMP/COMP-1/COMP-2, elementary `OCCURS n` (flattened to
/// `name_1..name_n`), and `REDEFINES` **overlays** (the redefining field is
/// placed at the redefined field's offset and does NOT advance the record
/// cursor, so both alternatives are present in the schema over the same bytes).
/// `FILLER` consumes bytes but is dropped from the schema.
/// **Seam**: group-level `OCCURS DEPENDING ON` is not expanded (documented
/// follow-up).
pub fn parse_copybook(text: &str) -> Result<FieldLayout> {
let entries = tokenize(text)?;
let mut fields = Vec::new();
let mut cursor = 0usize;
// Start offset of every placed entry (group + elementary), keyed by
// `(level, name)`, so a later `REDEFINES target` can overlay at `target`'s
// original offset. COBOL requires the subject share the redefining item's
// level — hence the level is part of the key.
let mut placed: Vec<(u8, String, usize)> = Vec::new();
for e in &entries {
// Group entry (no PIC) is organizational — record its start offset (so a
// group REDEFINES can find it) then let its children follow inline.
let Some(pic) = e.pic else {
placed.push((e.level, e.name.clone(), cursor));
continue;
};
let len = len_of(&pic, e.usage);
// A REDEFINES overlays the named subject at its offset without reserving
// new bytes; a normal field is placed at the cursor and advances it.
let (base_offset, advance) = match &e.redefines {
Some(target) => {
let off = placed
.iter()
.rev()
.find(|(lvl, name, _)| *lvl == e.level && name == target)
.or_else(|| placed.iter().rev().find(|(_, name, _)| name == target))
.map(|(_, _, off)| *off)
.unwrap_or(cursor);
(off, false)
}
None => (cursor, true),
};
placed.push((e.level, e.name.clone(), base_offset));
let is_filler = e.name.eq_ignore_ascii_case("FILLER");
for k in 0..e.occurs.max(1) {
let field_offset = base_offset + k * len;
if !is_filler {
let name = if e.occurs > 1 {
format!("{}_{}", e.name, k + 1)
} else {
e.name.clone()
};
fields.push(Field {
name,
offset: field_offset,
len,
usage: e.usage,
pic,
arrow_type: arrow_type_of(&pic, e.usage),
});
}
if advance {
cursor += len;
}
}
}
if fields.is_empty() {
return Err(anyhow!("copybook parsed to zero elementary fields"));
}
Ok(FieldLayout {
fields,
lrecl: cursor,
})
}
/// Split copybook text into raw entries. A statement ends at `.`; lines whose
/// first non-blank char is `*` (or COBOL indicator-area `*`/`/`) are comments.
fn tokenize(text: &str) -> Result<Vec<RawEntry>> {
// drop comment lines, then join and split on '.'
let cleaned: String = text
.lines()
.map(str::trim_end)
.filter(|l| {
let t = l.trim_start();
!(t.is_empty() || t.starts_with('*') || t.starts_with('/'))
})
.collect::<Vec<_>>()
.join(" ");
let mut out = Vec::new();
for stmt in cleaned.split('.') {
let toks: Vec<&str> = stmt.split_whitespace().collect();
if toks.is_empty() {
continue;
}
let Ok(level) = toks[0].parse::<u8>() else {
continue; // not a data-description entry
};
let name = toks.get(1).copied().unwrap_or("FILLER").to_string();
let mut pic = None;
let mut usage = Usage::Display;
let mut occurs = 1usize;
let mut redefines: Option<String> = None;
let mut i = 1;
while i < toks.len() {
let t = toks[i].to_ascii_uppercase();
match t.as_str() {
"PIC" | "PICTURE" => {
// optional "IS"
let mut j = i + 1;
if toks
.get(j)
.map(|s| s.eq_ignore_ascii_case("IS"))
.unwrap_or(false)
{
j += 1;
}
if let Some(p) = toks.get(j) {
pic = Some(parse_pic(p));
}
i = j;
}
"COMP-3" | "COMPUTATIONAL-3" | "PACKED-DECIMAL" => usage = Usage::Comp3,
"COMP" | "COMP-4" | "COMPUTATIONAL" | "COMPUTATIONAL-4" | "BINARY" => {
usage = Usage::Comp
}
"COMP-1" | "COMPUTATIONAL-1" => usage = Usage::Comp1,
"COMP-2" | "COMPUTATIONAL-2" => usage = Usage::Comp2,
"DISPLAY" => usage = Usage::Display,
"OCCURS" => {
if let Some(n) = toks.get(i + 1).and_then(|s| s.parse::<usize>().ok()) {
occurs = n;
i += 1;
}
}
"REDEFINES" => {
// the next token is the redefined (subject) field name
if let Some(target) = toks.get(i + 1) {
redefines = Some(target.to_string());
}
i += 1;
}
_ => {}
}
i += 1;
}
out.push(RawEntry {
level,
name,
pic,
usage,
occurs,
redefines,
});
}
Ok(out)
}
// ── ASCII fixed-width decode ──────────────────────────────────────────────────
/// Decode one ASCII/Latin-1 field slice → the field's Arrow value, appended to
/// the right builder. Numeric fields are read as ASCII digit text (USAGE DISPLAY
/// assumed — COMP-3/binary are EBCDIC-only in the mainframe path).
fn ascii_cell_text(slice: &[u8]) -> String {
// Latin-1: each byte is its own Unicode code point ≤ 0xFF.
slice.iter().map(|&b| b as char).collect::<String>()
}
/// Parse an ASCII numeric field into an unscaled i128 honouring the implied `V`
/// scale and a trailing/leading sign. Blank field → None.
fn ascii_number(slice: &[u8], pic: &PicInfo) -> Option<i128> {
let s = ascii_cell_text(slice);
let s = s.trim();
if s.is_empty() {
return None;
}
// strip an explicit decimal point / sign; keep digits, honour V scale.
let neg = s.starts_with('-');
let digits: String = s.chars().filter(|c| c.is_ascii_digit()).collect();
if digits.is_empty() {
return None;
}
let mut val: i128 = digits.parse().ok()?;
// if the text carried an explicit '.', its fractional digits are already in
// `val`; otherwise the field is an implied-decimal integer of `int+frac`
// digits, so `val` already IS the unscaled mantissa. Either way `val` is the
// unscaled integer over all digit chars — which matches scale = frac_digits.
let _ = pic;
if neg {
val = -val;
}
Some(val)
}
/// Column builders shared by the ASCII + EBCDIC record decoders.
pub(crate) enum ColBuilder {
Text(StringBuilder),
I32(Int32Builder),
I64(Int64Builder),
Dec(Decimal128Builder, u8, i8),
}
impl ColBuilder {
pub(crate) fn for_field(f: &Field) -> Self {
match &f.arrow_type {
DataType::Utf8 => ColBuilder::Text(StringBuilder::new()),
DataType::Int32 => ColBuilder::I32(Int32Builder::new()),
DataType::Int64 => ColBuilder::I64(Int64Builder::new()),
DataType::Decimal128(p, s) => ColBuilder::Dec(Decimal128Builder::new(), *p, *s),
other => panic!("unsupported mainframe arrow type {other:?}"),
}
}
pub(crate) fn push_text(&mut self, v: Option<String>) {
if let ColBuilder::Text(b) = self {
match v {
Some(s) => b.append_value(s),
None => b.append_null(),
}
}
}
pub(crate) fn push_int(&mut self, v: Option<i128>) {
match self {
ColBuilder::I32(b) => match v {
Some(n) => b.append_value(n as i32),
None => b.append_null(),
},
ColBuilder::I64(b) => match v {
Some(n) => b.append_value(n as i64),
None => b.append_null(),
},
ColBuilder::Dec(b, _, _) => match v {
Some(n) => b.append_value(n),
None => b.append_null(),
},
ColBuilder::Text(b) => match v {
Some(n) => b.append_value(n.to_string()),
None => b.append_null(),
},
}
}
pub(crate) fn finish(self) -> ArrayRef {
use std::sync::Arc;
match self {
ColBuilder::Text(mut b) => Arc::new(b.finish()),
ColBuilder::I32(mut b) => Arc::new(b.finish()),
ColBuilder::I64(mut b) => Arc::new(b.finish()),
ColBuilder::Dec(mut b, p, s) => Arc::new(
b.finish()
.with_precision_and_scale(p, s)
.expect("valid decimal"),
),
}
}
}
/// Slice `bytes` into fixed `LRECL` records (RECFM=FB). A trailing short chunk
/// (padding) is ignored.
pub fn slice_records(bytes: &[u8], lrecl: usize) -> Vec<&[u8]> {
if lrecl == 0 {
return vec![];
}
bytes.chunks_exact(lrecl).collect()
}
/// Decode an **ASCII/Latin-1** fixed-width blob against a copybook layout →
/// Arrow. Numeric fields are ASCII digits (implied-decimal via the PIC `V`
/// scale); text is Latin-1. (EBCDIC + COMP-3 live in [`crate::mainframe`].)
pub fn decode_ascii_records(layout: &FieldLayout, bytes: &[u8]) -> 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()) {
let slice = &rec[f.offset..f.offset + f.len];
if f.pic.is_text {
let s = ascii_cell_text(slice).trim_end().to_string();
b.push_text(Some(s));
} else {
b.push_int(ascii_number(slice, &f.pic));
}
}
}
let columns: Vec<ArrayRef> = builders.into_iter().map(ColBuilder::finish).collect();
Ok(RecordBatch::try_new(
std::sync::Arc::new(layout.schema()),
columns,
)?)
}
// ── ASCII fixed-width collector ───────────────────────────────────────────────
/// The generic (non-mainframe) ASCII fixed-width collector — same copybook
/// layout parser as [`crate::mainframe`], ASCII decode, no EBCDIC/COMP-3.
pub struct FixedWidthCollector {
gdpr: GdprClass,
}
impl Default for FixedWidthCollector {
fn default() -> Self {
Self {
gdpr: GdprClass::Personal,
}
}
}
fn copybook_from_cfg(cfg: &CollectorCfg) -> Result<FieldLayout> {
let text = cfg
.params
.get("copybook")
.and_then(|v| v.as_str())
.ok_or_else(|| {
anyhow!("mainframe/fixed-width needs `params.copybook` (COBOL copybook text)")
})?;
parse_copybook(text)
}
/// **The human-card half shared by every copybook-driven flat-file collector** —
/// [`FixedWidthCollector`] here and [`crate::mainframe::MainframeCollector`] next door.
///
/// Both read a file korp did not fetch, from a source korp cannot name: the operator
/// points `base_url` at a path (or an `http(s)` URL) and supplies the COBOL copybook
/// that describes it. There is no authority, no account, nothing to register and no
/// rate limit — and that is a REAL answer rather than a gap, which is why it is
/// written out in full instead of left as `Unknown`. Written once so the two cards
/// cannot drift into disagreeing about a fact neither of them owns (LAW 5).
///
/// `terms_url` is korp-collectors' own repository, and that is deliberately the
/// NARROWER claim: MIT-OR-Apache-2.0 covers the reader. The data's licence travels
/// with the file and only whoever supplied it can state it — the first quirk says so
/// before anything else, so nobody reads the repository licence as the data's.
pub(crate) fn supplied_file_card(docs_url: &str) -> crate::human_doc::HumanDoc {
use crate::human_doc::{Contact, RateLimit};
crate::human_doc::HumanDoc::new(
"no external authority — the operator supplies the file, so whoever issued the \
export is the authority and korp cannot name them",
"whoever supplied the file (and the COBOL copybook that describes it)",
"https://git.gunnar.rs/nordisk/korp-collectors",
docs_url,
)
.with_contact(Contact::NonePublished)
.with_rate_limit(RateLimit::Documented(
"none — the collector reads the whole file once per run. `base_url` may also be \
an `http(s)` URL, in which case the only limit is the hosting site's and korp \
does not know it."
.into(),
))
.with_quirk(
"THE DATA'S LICENCE TRAVELS WITH THE FILE. `terms_url` above is \
korp-collectors' own repository (MIT OR Apache-2.0) and covers the READER \
only — it says nothing about what you may do with the records. Ask whoever \
handed you the export; korp has no way to find out.",
)
.with_quirk(
"`params.copybook` is REQUIRED and there is no fallback: without it both \
`schema()` and `collect()` fail with `mainframe/fixed-width needs \
\\`params.copybook\\` (COBOL copybook text)`. The copybook IS the schema — \
nothing is inferred from the data, ever, which is the whole point of this \
collector.",
)
.with_quirk(
"LRECL is DERIVED from the copybook (the running sum of the field lengths), not \
configured. A copybook that omits FILLER bytes therefore produces a short \
LRECL, and every record after the first is read at the wrong offset — garbled \
columns, no error. Reconcile the derived LRECL against the dataset's real one \
before trusting a first run.",
)
.with_quirk(
"The slicer is `chunks_exact(LRECL)`: a trailing partial record — a truncated \
transfer, or the wrong LRECL — is dropped SILENTLY. A file shorter than one \
record slices to zero and comes back as a reason naming the LRECL and the byte \
count, which is the honest failure; the dangerous case is the wrong LRECL that \
still divides evenly.",
)
.with_quirk(
"RECFM=FB only. A variable-length (RECFM=V/VB) dataset carries a 4-byte RDW \
length prefix per record and there is no code path for it here — it will slice \
at the wrong boundaries rather than complain. Unload as fixed-blocked.",
)
.with_quirk(
"`base_url` is EMPTY by default (`SourceTarget::warehouse(FlatFile, \"\", …)`), \
so an unconfigured collector reports `Missing::Source` — \"no source URL is \
declared, so no request can be built\" — rather than looking for a file.",
)
}
#[async_trait::async_trait]
impl Collector for FixedWidthCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "fixed-width".into(),
direction: Direction::Source,
source_kind: SourceKind::FlatFile,
path: CollectorPath::RustNative,
formats: vec!["fixed-width".into(), "cobol-copybook".into()],
license: "per-source".into(),
gdpr: self.gdpr,
auth: AuthKind::None,
join_keys: vec![],
description: "ASCII/Latin-1 fixed-width flat file decoded against a COBOL \
copybook layout (DISPLAY usage; no EBCDIC/COMP-3)."
.into(),
}
}
/// The shared [`supplied_file_card`] plus the two facts specific to the ASCII arm:
/// what it will NOT decode, and the GDPR class it declares by default.
///
/// `docs_url` points at this module's own source, and that is the honest choice
/// rather than a cop-out: the copybook subset korp parses is defined HERE — the
/// `PIC`/`USAGE` forms in [`parse_copybook`] — not by any single published COBOL
/// standard page an operator could act on.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
Some(
supplied_file_card(
"https://git.gunnar.rs/nordisk/korp-collectors",
)
.with_quirk(
"ASCII/Latin-1 and `USAGE DISPLAY` ONLY. A copybook declaring COMP-3 \
(packed), COMP (big-endian binary) or zoned numerics, or a file in \
EBCDIC, belongs to `mainframe-copybook` — the two share this module's \
layout parser, so the same copybook works, but this arm's decode path \
will read those bytes as text.",
)
.with_quirk(
"GDPR class is `Personal`, not `Open` — the registered collector is built \
from `Default` and its `gdpr` field is private, so there is no way to \
relax it from configuration. Fixed-width exports are usually somebody's \
records; expect the warehouse to treat every row as personal data.",
),
)
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(SourceEndpoint::FlatFile, "", "fixedwidth_import")
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
Ok(copybook_from_cfg(cfg)?.schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let layout = copybook_from_cfg(cfg)?;
let bytes = crate::http::read_source(cfg, &cfg.base_url).await?;
let rb = decode_ascii_records(&layout, &bytes)?;
let rows = rb.num_rows() as u64;
if rows == 0 {
return Ok(CollectStats {
reason: Some(format!(
"fixed-width `{}` sliced to 0 records (LRECL={})",
cfg.base_url, layout.lrecl
)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}