#870 Add support for in-place processing of VRL files with RDWs

README.md

@@ -2165,12 +2165,12 @@ A: Update hadoop dll to version 3.2.2 or newer.
     - [#869](https://github.com/AbsaOSS/cobrix/pull/869) Added support for REDEFINES when writing EBCDIC files from Spark DataFrames. Thanks @Il-Pela!
     - [#757](https://github.com/AbsaOSS/cobrix/issues/757) Added support for GPG transparent decryption when reading data files
       ```scala
-      val df = spark.read
+      spark.read
         .format("cobol")
         .option("copybook", copybookPath)
         .option("gpg_private_key", gpgPrivateKey) // -----BEGIN PGP PRIVATE KEY BLOCK----- ...
-        .option("gpg_private_key", gpg_private_key_passphrase)
-        .load("/some/path")
+        .option("gpg_private_key_passphrase", gpg_private_key_passphrase)
+        .load("/some/path/*.gpg")
       ```
 
 - #### 2.10.8 released 3 August 2026.

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/headerparsers/RecordHeaderParserRDW.scala

@@ -25,6 +25,7 @@ class RecordHeaderParserRDW(isBigEndian: Boolean,
                             fileHeaderBytes: Int,
                             fileFooterBytes: Int,
                             rdwAdjustment: Int) extends Serializable with RecordHeaderParser {
+  import RecordHeaderParserRDW._
 
   /** RDW header is a 4 byte header */
   override def getHeaderLength: Int = 4
@@ -48,21 +49,23 @@ class RecordHeaderParserRDW(isBigEndian: Boolean,
     } else if (fileSize > 0L && fileFooterBytes > 0 && fileSize - fileOffset <= fileFooterBytes) {
       RecordMetadata((fileSize - fileOffset - fileFooterBytes).toInt, isValid = false)
     } else {
-      processRdwHeader(header, fileOffset)
+      processRdwHeader(header, fileOffset, isBigEndian, rdwAdjustment)
     }
   }
+}
 
+object RecordHeaderParserRDW {
   /**
     * Parses an RDW header.
     *
-    * @param header A record header as an array of bytes
-    * @param offset An offset from the beginning of the underlying file
-    *
+    * @param header        A record header as an array of bytes
+    * @param offset        An offset from the beginning of the underlying file
+    * @param isBigEndian   A flag indicating if the header is in big-endian format
+    * @param rdwAdjustment An adjustment value for the RDW header
     * @return A parsed record metadata
     */
-  private def processRdwHeader(header: Array[Byte], offset: Long): RecordMetadata = {
-    val rdwHeaderBlock = getHeaderLength
-    if (header.length < rdwHeaderBlock) {
+  def processRdwHeader(header: Array[Byte], offset: Long, isBigEndian: Boolean, rdwAdjustment: Int): RecordMetadata = {
+    if (header.length < 4) {
       RecordMetadata(-1, isValid = false)
     }
     else {
@@ -91,5 +94,4 @@ class RecordHeaderParserRDW(isBigEndian: Boolean,
       }
     }
   }
-
 }

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala

@@ -90,6 +90,8 @@ class VarLenNestedReader[T: ClassTag](copybookContents: Seq[String],
           Some(new FixedRecordLengthRawRecordExtractor(reParams, readerProperties.recordLength))
       case None if readerProperties.recordFormat == FixedLength   =>
         Some(new FixedRecordLengthRawRecordExtractor(reParams, readerProperties.recordLength))
+      case None if readerProperties.recordFormat == VariableLength && readerProperties.recordHeaderParser.isEmpty && readerProperties.isRecordSequence =>
+        Some(new VariableRecordLengthRecordExtractor(reParams))
       case None                                                   =>
         None
     }

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala

@@ -16,13 +16,33 @@
 
 package za.co.absa.cobrix.cobol.reader.extractors.raw
 
+/**
+  * A raw record extractor for files that consist of records of the same fixed length.
+  *
+  * Records are read sequentially from the input stream, each record consuming exactly the configured
+  * number of bytes. When the record length is not provided explicitly, it is derived from the copybook,
+  * so the extractor can be used for any file which layout implies a constant record size.
+  *
+  * The extraction stops as soon as the end of the input stream is reached, or when no more bytes can be
+  * fetched from it. The last chunk of data is returned as a record even if it is shorter than the expected
+  * record size.
+  *
+  * The header stream is not used by this extractor since no file level headers need to be inspected,
+  * therefore it is closed immediately on construction.
+  *
+  * @param ctx               A context of the record extractor containing the input stream, the copybook
+  *                          and the options passed to `spark-cobol`.
+  * @param fixedRecordLength An optional record length in bytes. If not specified, the record size
+  *                          calculated from the copybook is used.
+  */
 class FixedRecordLengthRawRecordExtractor(ctx: RawRecordContext, fixedRecordLength: Option[Int]) extends Serializable with RawRecordExtractor {
   private var byteOffset: Long = ctx.inputStream.offset
   private val recordSize = fixedRecordLength.getOrElse(ctx.copybook.getRecordSize)
   private var currentRecordOpt: Option[Array[Byte]] = None
 
   ctx.headerStream.close()
 
+  /** Returns the byte offset of the next record in the input stream. */
   override def offset: Long = byteOffset
 
   override def hasNext: Boolean = {
@@ -32,17 +52,7 @@ class FixedRecordLengthRawRecordExtractor(ctx: RawRecordContext, fixedRecordLeng
     currentRecordOpt.nonEmpty
   }
 
-  private def readNextRecord(): Unit = {
-    if (!ctx.inputStream.isEndOfStream) {
-      val nextRecord = ctx.inputStream.next(recordSize)
-
-      if (nextRecord.length > 0) {
-        currentRecordOpt = Some(nextRecord)
-      }
-    }
-  }
-
-
+  /** Returns the next record from the input stream. */
   @throws[NoSuchElementException]
   override def next(): Array[Byte] = {
     if (!hasNext) {
@@ -53,4 +63,15 @@ class FixedRecordLengthRawRecordExtractor(ctx: RawRecordContext, fixedRecordLeng
     currentRecordOpt = None
     record
   }
+
+  /** Reads the next record from the input stream if available. */
+  private def readNextRecord(): Unit = {
+    if (!ctx.inputStream.isEndOfStream) {
+      val nextRecord = ctx.inputStream.next(recordSize)
+
+      if (nextRecord.length > 0) {
+        currentRecordOpt = Some(nextRecord)
+      }
+    }
+  }
 }

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedWithRecordLengthExprRawRecordExtractor.scala

@@ -26,6 +26,31 @@ import za.co.absa.cobrix.cobol.utils.StringUtils
 
 import scala.util.Try
 
+/**
+  * A raw record extractor for files that consist of records having a variable length that is determined by
+  * the contents of the record itself rather than by an RDW header.
+  *
+  * The length of each record is resolved in one of the following ways, depending on the reader parameters:
+  *   - by reading a copybook field that contains the record length as a number,
+  *   - by reading a copybook field and translating its value into a record length using a value-to-length
+  *     mapping, where the key `"_"` can be used to define the default length for unmapped values,
+  *   - by evaluating an expression based on one or more copybook fields.
+  *
+  * The extractor reads only the leading portion of a record that is required to obtain the length (taking into
+  * account the configured start offset), then fetches the remaining bytes of the record, applying the end
+  * offset and the record length adjustment. Records that do not fall into the configured minimum and maximum
+  * record length range are skipped. When a segment id field is defined, its value is extracted for each record
+  * as well.
+  *
+  * The extractor reads records eagerly, one record ahead, so that the current offset in the underlying stream
+  * always points to the beginning of the record that has not been returned yet.
+  *
+  * @param ctx              the context of the raw record extraction containing the copybook, the input stream
+  *                         positioned at the first record, and the header stream, which is closed on construction
+  * @param readerProperties the parameters defining the record length field or expression, the length value
+  *                         mapping, the file start offset, record start and end offsets, minimum and maximum
+  *                         record lengths, the record length adjustment, and multisegment settings
+  */
 class FixedWithRecordLengthExprRawRecordExtractor(ctx: RawRecordContext,
                                                   readerProperties: ReaderParameters) extends Serializable with RawRecordExtractor {
   private val log = LoggerFactory.getLogger(this.getClass)

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/RawRecordContext.scala

@@ -51,19 +51,19 @@ object RawRecordContext {
       inputStream,
       headerStream,
       copybook,
-      new RecordHeaderDecoderRdw(RecordHeaderParameters(isBigEndian = true, 0)),
-      new RecordHeaderDecoderBdw(RecordHeaderParameters(isBigEndian = true, 0)),
+      new RecordHeaderDecoderRdw(RecordHeaderParameters(isBigEndian = true, 0, headersPartOfRecordLength = false)),
+      new RecordHeaderDecoderBdw(RecordHeaderParameters(isBigEndian = true, 0, headersPartOfRecordLength = false)),
       "",
       Map.empty[String, String]
     )
 
     def withReaderParams(readerParameters: ReaderParameters): RawRecordContextBuilder = {
-      val rdwParams = RecordHeaderParameters(readerParameters.isRdwBigEndian, readerParameters.rdwAdjustment)
+      val rdwParams = RecordHeaderParameters(readerParameters.isRdwBigEndian, readerParameters.rdwAdjustment, readerParameters.isRdwPartRecLength)
 
       val rdwDecoder = new RecordHeaderDecoderRdw(rdwParams)
 
       val bdwOpt = readerParameters.bdw
-      val bdwParamsOpt = bdwOpt.map(bdw => RecordHeaderParameters(bdw.isBigEndian, bdw.adjustment))
+      val bdwParamsOpt = bdwOpt.map(bdw => RecordHeaderParameters(bdw.isBigEndian, bdw.adjustment, headersPartOfRecordLength = false))
       val bdwDecoderOpt = bdwParamsOpt.map(bdwParams => new RecordHeaderDecoderBdw(bdwParams))
 
       withAdditionalInfo(readerParameters.reAdditionalInfo)

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractor.scala

@@ -18,24 +18,78 @@ package za.co.absa.cobrix.cobol.reader.extractors.raw
 
 import scala.collection.mutable
 
+/**
+  * A raw record extractor for variable block length files (`record_format = VB`), where records
+  * are grouped into blocks and both blocks and records are prefixed with their own headers.
+  *
+  * The extractor reads the input stream block by block. Each block begins with a block descriptor
+  * word (BDW), decoded by `ctx.bdwDecoder`, that defines the length of the block payload. The block
+  * payload, in turn, contains a sequence of records, each one prefixed with a record descriptor word
+  * (RDW), decoded by `ctx.rdwDecoder`, that defines the length of the record. All records of a block
+  * are buffered in an internal queue and are returned one by one by `next()`. Records with an empty
+  * payload are skipped. Once the queue is exhausted, the next block is fetched from the stream, and
+  * the iteration ends when the end of the stream is reached.
+  *
+  * Since block headers can only be interpreted at block boundaries, splitting the input is allowed
+  * only at offsets that point to the beginning of a block, which is reflected by `canSplitHere`.
+  * The offset returned by `offset` always points to the absolute beginning of the next record to be
+  * returned, including the BDW of the block it belongs to when that record is the first record of
+  * the block.
+  *
+  * The header stream is not used by this extractor and is closed on construction.
+  *
+  * @param ctx a context that holds the input stream, the copybook, the block and record header
+  *            decoders and the options passed to `spark-cobol`.
+  */
 class VariableBlockVariableRecordExtractor(ctx: RawRecordContext) extends Serializable with RawRecordExtractor {
   ctx.headerStream.close()
 
   private val recordQueue = new mutable.Queue[Array[Byte]]
   private var canSplitAtCurrentOffset = true
   private var recordOffset: Long = ctx.inputStream.offset
 
+  /** Returns the byte offset of the next record in the input stream. */
   override def offset: Long = recordOffset
 
+  /** Returns true if the input stream can be split at the current offset. */
   override def canSplitHere: Boolean = canSplitAtCurrentOffset
 
+  /** Returns true if there are more records to be read from the input stream. */
   override def hasNext: Boolean = {
     if (recordQueue.isEmpty) {
       readNextBlock()
     }
     recordQueue.nonEmpty
   }
 
+  /** Returns the next record from the input stream. */
+  @throws[NoSuchElementException]
+  override def next(): Array[Byte] = {
+    if (!hasNext) {
+      throw new NoSuchElementException
+    }
+    if (canSplitAtCurrentOffset) {
+      recordOffset += ctx.bdwDecoder.headerSize
+    }
+    val record = recordQueue.dequeue()
+    recordOffset += ctx.rdwDecoder.headerSize + record.length
+
+    canSplitAtCurrentOffset = recordQueue.isEmpty
+    record
+  }
+
+  /**
+    * Reads the next block (BDW) from the input stream and splits it into individual records.
+    *
+    * The block descriptor word is decoded first in order to determine the length of the block payload.
+    * The payload is then traversed record by record: for each record the record descriptor word is decoded
+    * to get the record length, and the corresponding non-empty payload is put into the internal record queue,
+    * from which subsequent records are served.
+    *
+    * If the end of the input stream has been reached, nothing is read and the record queue stays unchanged.
+    *
+    * @return nothing, the decoded records are added to the internal record queue as a side effect
+    */
   private def readNextBlock(): Unit = {
     val bdwSize = ctx.bdwDecoder.headerSize
     val rdwSize = ctx.rdwDecoder.headerSize
@@ -62,20 +116,4 @@ class VariableBlockVariableRecordExtractor(ctx: RawRecordContext) extends Serial
       }
     }
   }
-
-
-  @throws[NoSuchElementException]
-  override def next(): Array[Byte] = {
-    if (!hasNext) {
-      throw new NoSuchElementException
-    }
-    if (canSplitAtCurrentOffset) {
-      recordOffset += ctx.bdwDecoder.headerSize
-    }
-    val record = recordQueue.dequeue()
-    recordOffset += ctx.rdwDecoder.headerSize + record.length
-
-    canSplitAtCurrentOffset = recordQueue.isEmpty
-    record
-  }
 }