Skip to content

Add support for writing VB (variable block) COBOL files #867

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yruslan merged 1 commit into from
Aug 4, 2026
Merged

Add support for writing VB (variable block) COBOL files #867

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions ...rser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,7 @@ object CobolParametersParser extends Logging {

val paths = pathsParam.map(_.split(',')).getOrElse(Array(getParameter(PARAM_SOURCE_PATH, params).getOrElse("")))

val variableLengthParams = parseVariableLengthParameters(params, recordFormatDefined)
val variableLengthParams = parseVariableLengthParameters(params, recordFormatDefined, isWriter)

val recordFormat = if (recordFormatDefined == AsciiText && variableLengthParams.nonEmpty) {
logger.info("According to options passed, the custom ASCII parser (record_format = D2) is used.")
Expand Down Expand Up @@ -551,7 +551,7 @@ object CobolParametersParser extends Logging {
)
}

private def parseVariableLengthParameters(params: Parameters, recordFormat: RecordFormat): Option[VariableLengthParameters] = {
private def parseVariableLengthParameters(params: Parameters, recordFormat: RecordFormat, isWriter: Boolean): Option[VariableLengthParameters] = {
val recordLengthFieldOpt = params.get(PARAM_RECORD_LENGTH_FIELD)
val isRecordSequence = Seq(FixedBlock, VariableLength, VariableBlock).contains(recordFormat)
val isRecordIdGenerationEnabled = params.getOrElse(PARAM_GENERATE_RECORD_ID, "false").toBoolean
Expand Down Expand Up @@ -598,7 +598,7 @@ object CobolParametersParser extends Logging {
Some(VariableLengthParameters
(
isRecordSequence,
parseBdw(params, recordFormat),
parseBdw(params, recordFormat, isWriter),
params.getOrElse(PARAM_IS_RDW_BIG_ENDIAN, "false").toBoolean,
params.getOrElse(PARAM_IS_RDW_PART_REC_LENGTH, "false").toBoolean,
params.getOrElse(PARAM_RDW_ADJUSTMENT, "0").toInt,
Expand Down Expand Up @@ -626,7 +626,7 @@ object CobolParametersParser extends Logging {
}
}

private def parseBdw(params: Parameters, recordFormat: RecordFormat): Option[Bdw] = {
private def parseBdw(params: Parameters, recordFormat: RecordFormat, isWriter: Boolean): Option[Bdw] = {
if (recordFormat == FixedBlock || recordFormat == VariableBlock) {
val bdw = Bdw(
params.getOrElse(PARAM_IS_BDW_BIG_ENDIAN, "false").toBoolean,
Expand All @@ -637,7 +637,9 @@ object CobolParametersParser extends Logging {
if (bdw.blockLength.nonEmpty && bdw.recordsPerBlock.nonEmpty) {
throw new IllegalArgumentException(s"Options '$PARAM_BLOCK_LENGTH' and '$PARAM_RECORDS_PER_BLOCK' cannot be used together.")
}
if (recordFormat == VariableBlock && bdw.blockLength.nonEmpty) {
// When reading VB the block length is self-described by the BDW header in the file, so 'block_length' is ignored.
// When writing VB, 'block_length' is meaningful: it caps how many records are packed into each block.
if (recordFormat == VariableBlock && bdw.blockLength.nonEmpty && !isWriter) {
logger.warn(s"Option '$PARAM_BLOCK_LENGTH' is ignored for record format: VB")
}
Comment on lines +640 to 644

@coderabbitai coderabbitai Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Look for documentation of VB writing and the blocking options.
fd -e md . | xargs rg -n -C3 'records_per_block|block_length|record_format.{0,10}VB'

Repository: AbsaOSS/cobrix

Length of output: 6603

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== locate relevant files =="
git ls-files | rg '(^README\.md$|cobol-parser/src/main/scala/.*/CobolParametersParser\.scala$|writer|cobol/writer|output)' | sed -n '1,200p'

echo
echo "== reader parser block_length warning context =="
file=$(git ls-files | rg 'cobol-parser/src/main/scala/.*/CobolParametersParser\.scala$' | head -n1)
echo "FILE=$file"
if [ -n "${file:-}" ]; then
  wc -l "$file"
  sed -n '600,660p' "$file" | nl -ba -v600
fi

echo
echo "== options table lines 1740-1775 =="
sed -n '1740,1775p' README.md | nl -ba -v1740

echo
echo "== search variable block writing mentions with broad text =="
rg -n -C3 'variable[block ]+length|VB|records_per_block|block_length|writer|write|output' README.md cobol-parser/src/main/scala za || true

Repository: AbsaOSS/cobrix

Length of output: 1503

🏁 Script executed:

#!/bin/bash
set -euo pipefail

file="cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala"

echo "== reader parser block_length warning context =="
wc -l "$file"
sed -n '615,650p' "$file"

echo
echo "== write path references to block_length/records_per_block/VariableBlock =="
rg -n -C4 'blockLength|recordsPerBlock|block_length|records_per_block|VariableBlock|block_length|records_per_block|records_per_block|block_length' spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer README.md | sed -n '1,240p'

echo
echo "== options table lines 1740-1775 =="
sed -n '1740,1775p' README.md

echo
echo "== sample variable-block test names/descriptions =="
rg -n -C3 'variable|VB|blockLength|recordsPerBlock|records_per_block|block_length' spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableBlockEbcdicWriterSuite.scala | sed -n '1,220p'

Repository: AbsaOSS/cobrix

Length of output: 32804

Document the VB writer blocking behavior.

block_length and records_per_block both drive VB output, and both are required together for reading and exclusion, but README.md still lists them only under FB and omits the variable-record VB section. Update the options table/example text so users know either option is accepted for record_format = VB and only one may be specified.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala`
around lines 640 - 644, Update the README options table and example text for
variable-block (VB) records to document that block_length and records_per_block
are accepted for record_format = VB, must be provided together where applicable,
and are mutually exclusive so only one may be specified.

if (recordFormat == FixedBlock && bdw.recordsPerBlock.nonEmpty) {
Expand Down
14 changes: 12 additions & 2 deletions ...main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,11 +145,21 @@ object CobolParametersValidator {
def validateParametersForWriting(readerParameters: ReaderParameters): Unit = {
val issues = new ListBuffer[String]

if (readerParameters.recordFormat != RecordFormat.FixedLength && readerParameters.recordFormat != RecordFormat.VariableLength) {
issues += s"Only '${RecordFormat.FixedLength}' and '${RecordFormat.VariableLength}' values for 'record_format' are supported for writing, " +
if (readerParameters.recordFormat != RecordFormat.FixedLength &&
readerParameters.recordFormat != RecordFormat.VariableLength &&
readerParameters.recordFormat != RecordFormat.VariableBlock) {
issues += s"Only '${RecordFormat.FixedLength}', '${RecordFormat.VariableLength}' and '${RecordFormat.VariableBlock}' values for 'record_format' are supported for writing, " +
s"provided value: '${readerParameters.recordFormat}'"
}

if (readerParameters.recordFormat == RecordFormat.VariableBlock) {
val hasBlockLength = readerParameters.bdw.exists(_.blockLength.nonEmpty)
val hasRecordsPerBlock = readerParameters.bdw.exists(_.recordsPerBlock.nonEmpty)
if (!hasBlockLength && !hasRecordsPerBlock) {
issues += "Writing 'VB' records requires either 'records_per_block' or 'block_length' to be specified"
}
Comment on lines +155 to +160

@coderabbitai coderabbitai Bot Aug 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Reject non-positive VB blocking values.

This branch checks only option presence. It accepts block_length <= 0 and records_per_block <= 0. The parser converts these options directly to Int without a range check. (raw.githubusercontent.com) The downstream combiner treats records_per_block as an exact count, so these values do not describe a valid block configuration. (raw.githubusercontent.com)

Validate each defined value as positive and add tests for zero and negative values.

Proposed fix 
     if (readerParameters.recordFormat == RecordFormat.VariableBlock) {
+      readerParameters.bdw.foreach { bdw =>
+        bdw.blockLength.foreach { value =>
+          if (value <= 0) {
+            issues += "'block_length' must be a positive integer"
+          }
+        }
+        bdw.recordsPerBlock.foreach { value =>
+          if (value <= 0) {
+            issues += "'records_per_block' must be a positive integer"
+          }
+        }
+      }
       val hasBlockLength = readerParameters.bdw.exists(_.blockLength.nonEmpty)
       val hasRecordsPerBlock = readerParameters.bdw.exists(_.recordsPerBlock.nonEmpty)
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if (readerParameters.recordFormat == RecordFormat.VariableBlock) {
val hasBlockLength = readerParameters.bdw.exists(_.blockLength.nonEmpty)
val hasRecordsPerBlock = readerParameters.bdw.exists(_.recordsPerBlock.nonEmpty)
if (!hasBlockLength && !hasRecordsPerBlock) {
issues += "Writing 'VB' records requires either 'records_per_block' or 'block_length' to be specified"
}
if (readerParameters.recordFormat == RecordFormat.VariableBlock) {
readerParameters.bdw.foreach { bdw =>
bdw.blockLength.foreach { value =>
if (value <= 0) {
issues += "'block_length' must be a positive integer"
}
}
bdw.recordsPerBlock.foreach { value =>
if (value <= 0) {
issues += "'records_per_block' must be a positive integer"
}
}
}
val hasBlockLength = readerParameters.bdw.exists(_.blockLength.nonEmpty)
val hasRecordsPerBlock = readerParameters.bdw.exists(_.recordsPerBlock.nonEmpty)
if (!hasBlockLength && !hasRecordsPerBlock) {
issues += "Writing 'VB' records requires either 'records_per_block' or 'block_length' to be specified"
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala`
around lines 155 - 160, Update the VariableBlock validation in
CobolParametersValidator to reject any defined bdw.blockLength or
bdw.recordsPerBlock value that is zero or negative, while preserving the
existing requirement that at least one option is specified. Add tests covering
zero and negative values for both blocking options.

}

if (readerParameters.occursMappings.nonEmpty) {
issues += "OCCURS mapping option ('occurs_mappings') is not supported for writing"
}
Expand Down
124 changes: 122 additions & 2 deletions spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer/NestedRecordCombiner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,8 @@ class NestedRecordCombiner extends RecordCombiner {
* @return The RDD of records in mainframe format
*/
override def combine(df: DataFrame, cobolSchema: CobolSchema, readerParameters: ReaderParameters): RDD[Array[Byte]] = {
val hasRdw = readerParameters.recordFormat == RecordFormat.VariableLength
val isVb = readerParameters.recordFormat == RecordFormat.VariableBlock
val hasRdw = readerParameters.recordFormat == RecordFormat.VariableLength || isVb
val isRdwBigEndian = readerParameters.isRdwBigEndian
val adjustment1 = if (readerParameters.isRdwPartRecLength) 4 else 0
val adjustment2 = readerParameters.rdwAdjustment
Expand Down Expand Up @@ -73,7 +74,7 @@ class NestedRecordCombiner extends RecordCombiner {
s"RDW length $recordLengthLong exceeds ${Int.MaxValue} and cannot be encoded safely."
)
}
processRDD(df.rdd,
val recordsRdd = processRDD(df.rdd,
cobolSchema.copybook,
df.schema,
size,
Expand All @@ -84,12 +85,131 @@ class NestedRecordCombiner extends RecordCombiner {
readerParameters.variableSizeOccurs,
readerParameters.strictSchema,
readerParameters.writerParameters.get)

if (isVb) {
val bdw = readerParameters.bdw.getOrElse(
throw new IllegalArgumentException("Writing 'VB' records requires either 'records_per_block' or 'block_length' to be specified.")
)
groupIntoBlocks(recordsRdd, bdw.isBigEndian, bdw.adjustment, bdw.blockLength, bdw.recordsPerBlock)
} else {
recordsRdd
}
}

/**
* Groups an RDD of already-encoded records (each containing its own RDW + payload) into VB blocks.
*
* Each block is a single byte array of the form: BDW(4 bytes) ++ record1 ++ record2 ++ ...
* Grouping happens independently within each Spark partition (blocks never span partitions, which is
* consistent with the one-output-file-per-partition behaviour of the writer).
*
* Exactly one of `blockLength` / `recordsPerBlock` is expected to be set (mutual exclusion is enforced
* upstream during parameter parsing):
* - `recordsPerBlock`: put exactly N records per block; the last block of a partition may be smaller.
* - `blockLength`: pack records while the cumulative record bytes stay within the cap. A single record
* that is larger than the cap still gets its own block (records are never split).
*
* @param rdd The RDD of encoded records (RDW + payload each)
* @param isBdwBigEndian Byte order of the BDW header
* @param bdwAdjustment Value added to the block length before it is encoded into the BDW
* @param blockLength Optional cap (in bytes) on the cumulative record bytes per block
* @param recordsPerBlock Optional exact number of records per block
* @return An RDD where each element is one complete VB block (BDW + records)
*/
private[cobrix] def groupIntoBlocks(rdd: RDD[Array[Byte]],
isBdwBigEndian: Boolean,
bdwAdjustment: Int,
blockLength: Option[Int],
recordsPerBlock: Option[Int]): RDD[Array[Byte]] = {
rdd.mapPartitions { rawRecords =>
val records = rawRecords.buffered

new Iterator[Array[Byte]] {
override def hasNext: Boolean = records.hasNext

override def next(): Array[Byte] = {
val blockRecords = new mutable.ArrayBuffer[Array[Byte]]()

// Always take at least one record, so a record larger than 'block_length' still forms its own block.
val first = records.next()
blockRecords += first
var blockPayloadSize = first.length

var continue = true
while (continue && records.hasNext) {
val canAddByCount = recordsPerBlock.forall(n => blockRecords.length < n)
val canAddBySize = blockLength.forall(cap => blockPayloadSize + records.head.length <= cap)
if (canAddByCount && canAddBySize) {
val rec = records.next()
blockRecords += rec
blockPayloadSize += rec.length
} else {
continue = false
}
}

buildBlock(blockRecords, blockPayloadSize, isBdwBigEndian, bdwAdjustment)
}
}
}
}
}

object NestedRecordCombiner {
private val log = LoggerFactory.getLogger(this.getClass)

// The BDW normal form reserves the top bit as an "extended format" flag, leaving 15 bits for the length.
private final val MAX_BDW_BLOCK_LENGTH = 0x7FFF

/**
* Assembles a single VB block byte array from its records and prepends the 4-byte BDW header.
*
* @param records The encoded records (each already RDW + payload) that belong to this block
* @param blockPayloadSize The total number of bytes occupied by all records in the block (excludes the BDW)
* @param isBdwBigEndian Byte order of the BDW header
* @param bdwAdjustment Value added to the block length before encoding it into the BDW
* @return The complete block: BDW(4 bytes) followed by every record
*/
private[cobrix] def buildBlock(records: Iterable[Array[Byte]],
blockPayloadSize: Int,
isBdwBigEndian: Boolean,
bdwAdjustment: Int): Array[Byte] = {
val blockLengthField = blockPayloadSize + bdwAdjustment

if (blockLengthField < 0) {
throw new IllegalArgumentException(s"Invalid BDW block length $blockLengthField. Check 'bdw_adjustment'.")
}
if (blockLengthField > MAX_BDW_BLOCK_LENGTH) {
throw new IllegalArgumentException(
s"BDW block length $blockLengthField exceeds $MAX_BDW_BLOCK_LENGTH and cannot be encoded. Reduce 'block_length' or 'records_per_block'."
)
}

val block = new Array[Byte](4 + blockPayloadSize)

if (isBdwBigEndian) {
block(0) = ((blockLengthField >> 8) & 0x7F).toByte
block(1) = (blockLengthField & 0xFF).toByte
// The last two bytes are reserved and defined by IBM as binary zeros.
block(2) = 0
block(3) = 0
} else {
// Little-endian BDW: the first two bytes are reserved zeros, the length lives in bytes 2 and 3.
block(0) = 0
block(1) = 0
block(2) = (blockLengthField & 0xFF).toByte
block(3) = ((blockLengthField >> 8) & 0x7F).toByte
}

var offset = 4
records.foreach { rec =>
System.arraycopy(rec, 0, block, offset, rec.length)
offset += rec.length
}

block
}

/**
* Generates a field definition string containing the PIC clause and USAGE clause for a primitive COBOL field.
*
Expand Down
2 changes: 1 addition & 1 deletion ...scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidatorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ class CobolParametersValidatorSuite extends AnyWordSpec {
CobolParametersValidator.validateParametersForWriting(readParams)
}

assert(ex.getMessage.contains("Writer validation issues: Only 'F' and 'V' values for 'record_format' are supported for writing, provided value: 'VB';"))
assert(ex.getMessage.contains("Writing 'VB' records requires either 'records_per_block' or 'block_length' to be specified"))
assert(ex.getMessage.contains("OCCURS mapping option ('occurs_mappings') is not supported for writing"))
assert(ex.getMessage.contains("'record_start_offset' and 'record_end_offset' are not supported for writing"))
assert(ex.getMessage.contains("'file_start_offset' and 'file_end_offset' are not supported for writing"))
Expand Down
Loading
Loading