diff --git a/README.md b/README.md
index 504c0884c..f98047b71 100644
--- a/README.md
+++ b/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.
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/headerparsers/RecordHeaderParserRDW.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/headerparsers/RecordHeaderParserRDW.scala
index c869c3d0a..fba7f7094 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/parser/headerparsers/RecordHeaderParserRDW.scala
+++ b/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,
}
}
}
-
}
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala
index 556b83a06..a15f3587a 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala
+++ b/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
}
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala
index 2c0e79458..5457dc3a7 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala
+++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedRecordLengthRawRecordExtractor.scala
@@ -16,6 +16,25 @@
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)
@@ -23,6 +42,7 @@ class FixedRecordLengthRawRecordExtractor(ctx: RawRecordContext, fixedRecordLeng
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)
+ }
+ }
+ }
}
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedWithRecordLengthExprRawRecordExtractor.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedWithRecordLengthExprRawRecordExtractor.scala
index d4270bba8..ba1e5fb23 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedWithRecordLengthExprRawRecordExtractor.scala
+++ b/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)
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/RawRecordContext.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/RawRecordContext.scala
index fdec38ac6..c381743e8 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/RawRecordContext.scala
+++ b/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)
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractor.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractor.scala
index bea7977ae..b0bda3f42 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractor.scala
+++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableBlockVariableRecordExtractor.scala
@@ -18,6 +18,29 @@ 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()
@@ -25,10 +48,13 @@ class VariableBlockVariableRecordExtractor(ctx: RawRecordContext) extends Serial
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()
@@ -36,6 +62,34 @@ class VariableBlockVariableRecordExtractor(ctx: RawRecordContext) extends Serial
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
- }
}
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableRecordLengthRecordExtractor.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableRecordLengthRecordExtractor.scala
new file mode 100644
index 000000000..5580df228
--- /dev/null
+++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableRecordLengthRecordExtractor.scala
@@ -0,0 +1,115 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.cobrix.cobol.reader.extractors.raw
+
+import za.co.absa.cobrix.cobol.parser.headerparsers.RecordHeaderParserRDW
+import za.co.absa.cobrix.cobol.reader.recordheader.RecordHeaderDecoderRdw
+
+
+/**
+ * A [[RawRecordExtractor]] that reads variable-length records from an input stream
+ * using RDW (Record Descriptor Word) headers.
+ *
+ * Each record is preceded by a 4-byte RDW header that encodes the length of the
+ * following payload. Invalid or zero-length RDW headers are skipped until a valid
+ * record is found or the stream is exhausted.
+ *
+ * The header stream is closed immediately on construction since this extractor only
+ * needs to read forward through the input stream.
+ *
+ * @param ctx The raw record context providing the input stream, RDW decoder, and other
+ * reading parameters.
+ */
+class VariableRecordLengthRecordExtractor(ctx: RawRecordContext) extends Serializable with RawRecordExtractor {
+ ctx.headerStream.close()
+
+ private var currentRecord: Option[Array[Byte]] = None
+ private var recordOffset: Long = ctx.inputStream.offset
+ private val rdwParams = ctx.rdwDecoder.asInstanceOf[RecordHeaderDecoderRdw].rdwParameters
+ private val rdwAdjustment = if (rdwParams.headersPartOfRecordLength) rdwParams.adjustment - 4 else rdwParams.adjustment
+
+ /** Returns the byte offset of the next record in the input stream. */
+ override def offset: Long = recordOffset
+
+ /** Always returns `true`, since variable-length RDW records can always be split at record boundaries. */
+ override def canSplitHere: Boolean = true
+
+ /**
+ * Returns `true` if there is at least one more record available in the input stream.
+ *
+ * If no record has been pre-fetched yet, this method attempts to read the next record
+ * from the stream before returning.
+ */
+ override def hasNext: Boolean = {
+ if (currentRecord.isEmpty) {
+ readNextRecord()
+ }
+ currentRecord.nonEmpty
+ }
+
+
+ /**
+ * Returns the next raw record as an array of bytes and advances the stream position.
+ *
+ * Updates [[recordOffset]] to reflect the position of the record that will be returned
+ * by the subsequent call to `next()`.
+ *
+ * @throws NoSuchElementException if there are no more records in the stream.
+ * @return The raw bytes of the next record (excluding the RDW header).
+ */
+ @throws[NoSuchElementException]
+ override def next(): Array[Byte] = {
+ if (!hasNext) {
+ throw new NoSuchElementException
+ }
+
+ val record = currentRecord.get
+ recordOffset += ctx.rdwDecoder.headerSize + record.length
+
+ currentRecord = None
+ record
+ }
+
+ /**
+ * Reads the next record from the input stream by parsing the RDW header and extracting
+ * the subsequent payload.
+ *
+ * Skips over RDW headers that are marked as invalid. Stops when a valid record is found
+ * or the end of stream is reached.
+ */
+ private def readNextRecord(): Unit = {
+ val rdwSize = ctx.rdwDecoder.headerSize
+
+ var valid = false
+
+ while (!valid && !ctx.inputStream.isEndOfStream) {
+ val rdwOffset = ctx.inputStream.offset
+ val rdw = ctx.inputStream.next(rdwSize)
+
+ val m = RecordHeaderParserRDW.processRdwHeader(rdw, rdwOffset, rdwParams.isBigEndian, rdwAdjustment)
+ valid = m.isValid
+
+ if (m.recordLength > 0) {
+ val payload = ctx.inputStream.next(m.recordLength)
+
+ if (payload.length > 0) {
+ currentRecord = Some(payload)
+ }
+ }
+ }
+ }
+}
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderDecoderRdw.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderDecoderRdw.scala
index 42273722a..3b680eea4 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderDecoderRdw.scala
+++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderDecoderRdw.scala
@@ -20,7 +20,7 @@ package za.co.absa.cobrix.cobol.reader.recordheader
* This class represent a header decoder for standard RDW headers
* according to: https://www.ibm.com/docs/en/zos/2.3.0?topic=records-record-descriptor-word-rdw
*/
-class RecordHeaderDecoderRdw(rdwParameters: RecordHeaderParameters) extends RecordHeaderDecoderCommon {
+class RecordHeaderDecoderRdw(val rdwParameters: RecordHeaderParameters) extends RecordHeaderDecoderCommon {
final val RDW_HEADER_LENGTH = 4
override def headerSize: Int = RDW_HEADER_LENGTH
diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParameters.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParameters.scala
index 25dd0db05..bd1afa073 100644
--- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParameters.scala
+++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParameters.scala
@@ -22,5 +22,7 @@ case class RecordHeaderParameters(
/* Sometime the size includes only payload, and sometimes it includes headers themselves.
* This allows flexible adjustments. */
- adjustment: Int
+ adjustment: Int,
+
+ headersPartOfRecordLength: Boolean
)
diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorBaseSuite.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorBaseSuite.scala
index 143061cef..24eedb137 100644
--- a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorBaseSuite.scala
+++ b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/processor/impl/CobolProcessorBaseSuite.scala
@@ -19,8 +19,7 @@ package za.co.absa.cobrix.cobol.processor.impl
import org.scalatest.wordspec.AnyWordSpec
import za.co.absa.cobrix.cobol.mock.ByteStreamMock
import za.co.absa.cobrix.cobol.parser.recordformats.RecordFormat
-import za.co.absa.cobrix.cobol.processor.CobolProcessor
-import za.co.absa.cobrix.cobol.reader.extractors.raw.{FixedRecordLengthRawRecordExtractor, TextFullRecordExtractor}
+import za.co.absa.cobrix.cobol.reader.extractors.raw.{FixedRecordLengthRawRecordExtractor, TextFullRecordExtractor, VariableRecordLengthRecordExtractor}
import za.co.absa.cobrix.cobol.reader.parameters.ReaderParameters
class CobolProcessorBaseSuite extends AnyWordSpec {
@@ -43,7 +42,27 @@ class CobolProcessorBaseSuite extends AnyWordSpec {
assert(!ext.hasNext)
}
- "work for an variable-record-length files" in {
+ "work for an variable-record-length files with RDWs" in {
+ val stream = new ByteStreamMock(Array(
+ 0x02, 0x00, 0x00, 0x00, 0xF1, 0xF2,
+ 0x02, 0x00, 0x00, 0x00, 0xF3, 0xF4).map(_.toByte))
+
+ val ext = CobolProcessorBase.getRecordExtractor(
+ ReaderParameters(
+ recordFormat = RecordFormat.VariableLength,
+ isRecordSequence = true
+ ), copybook, stream, None
+ )
+
+ assert(ext.isInstanceOf[VariableRecordLengthRecordExtractor])
+
+ assert(ext.hasNext)
+ assert(ext.next().sameElements(Array(0xF1, 0xF2).map(_.toByte)))
+ assert(ext.next().sameElements(Array(0xF3, 0xF4).map(_.toByte)))
+ assert(!ext.hasNext)
+ }
+
+ "work for an variable-record-length text files" in {
val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
val ext = CobolProcessorBase.getRecordExtractor(ReaderParameters(
@@ -53,18 +72,5 @@ class CobolProcessorBaseSuite extends AnyWordSpec {
assert(ext.isInstanceOf[TextFullRecordExtractor])
}
-
- "throw an exception on a non-supported record format for processing" in {
- val stream = new ByteStreamMock(Array(0xF1, 0xF2, 0xF3, 0xF4).map(_.toByte))
-
- val ex = intercept[IllegalArgumentException] {
- CobolProcessorBase.getRecordExtractor(ReaderParameters(
- recordFormat = RecordFormat.VariableLength,
- isRecordSequence = true
- ), copybook, stream, None)
- }
-
- assert(ex.getMessage.contains("Cannot create a record extractor for the given reader parameters."))
- }
}
}
diff --git a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParametersFactory.scala b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParametersFactory.scala
index 1b3442390..971c69f9b 100644
--- a/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParametersFactory.scala
+++ b/cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/recordheader/RecordHeaderParametersFactory.scala
@@ -18,7 +18,8 @@ package za.co.absa.cobrix.cobol.reader.recordheader
object RecordHeaderParametersFactory {
def getDummyRecordHeaderParameters(isBigEndian: Boolean = false,
- adjustment: Int = 0): RecordHeaderParameters = {
- RecordHeaderParameters(isBigEndian, adjustment)
+ adjustment: Int = 0,
+ headersPartOfRecordLength: Boolean = false): RecordHeaderParameters = {
+ RecordHeaderParameters(isBigEndian, adjustment, headersPartOfRecordLength)
}
}
diff --git a/pom.xml b/pom.xml
index 91d097df8..52b35e450 100644
--- a/pom.xml
+++ b/pom.xml
@@ -105,7 +105,7 @@
2.12.21
2.12
3.5.7
- 3.2.19
+ 3.2.20
2.4.16
2.15.4
4.11.0
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 073ba5b27..a3cbe229d 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -23,7 +23,7 @@ object Dependencies {
private val jacksonVersion = "2.15.4"
private val bouncycastleVersion = "1.84"
- private val scalatestVersion = "3.2.19"
+ private val scalatestVersion = "3.2.20"
private val mockitoVersion = "4.11.0"
private val defaultSparkVersionForScala211 = "2.4.8"
diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala
index 9d48345ef..131ee3670 100644
--- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala
+++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/SparkCobolProcessorSuite.scala
@@ -149,6 +149,106 @@ class SparkCobolProcessorSuite extends AnyWordSpec with SparkTestBase with Binar
}
}
+ "convert from VRL+RDW into VRL+RDW simple" in {
+ val expected = """{"T":"0"}{"T":"1"}{"T":"2"}{"T":"3"}"""
+ withTempDirectory("spark_cobol_processor") { tempDir =>
+ val binData = Array(
+ 0x00, 0x00, 0x02, 0x00, 0xF1,
+ 0x00, 0x00, 0x02, 0x00, 0xF2,
+ 0x00, 0x00, 0x02, 0x00, 0xF3,
+ 0x00, 0x00, 0x02, 0x00, 0xF4).map(_.toByte)
+
+ val inputPath = new Path(tempDir, "input.dat").toString
+ val outputPath = new Path(tempDir, "output").toString
+ val outputFile = new Path(outputPath, "input.dat").toString
+
+ writeBinaryFile(inputPath, binData)
+
+ SparkCobolProcessor.builder
+ .withCopybookContents(copybook)
+ .option("record_format", "V")
+ .option("rdw_adjustment", "-1")
+ .option("is_rdw_big_endian", "false")
+ .withProcessingStrategy(CobolProcessingStrategy.ToVariableLength)
+ .withRecordProcessor(new SerializableRawRecordProcessor {
+ override def processRecord(record: Array[Byte], ctx: CobolProcessorContext): Array[Byte] = {
+ record.map(v => (v - 1).toByte)
+ }
+ })
+ .load(inputPath)
+ .save(outputPath)
+
+ val outputData = readBinaryFile(outputFile)
+
+ assert(outputData.sameElements(
+ Array(0, 1, 0, 0, -16, 0, 1, 0, 0, -15, 0, 1, 0, 0, -14, 0, 1, 0, 0, -13).map(_.toByte)
+ ))
+
+ val actual = spark.read
+ .format("cobol")
+ .option("copybook_contents", copybook)
+ .option("record_format", "V")
+ .option("is_rdw_big_endian", "true")
+ .option("pedantic", "true")
+ .load(outputFile)
+ .toJSON
+ .collect()
+ .mkString
+
+ assert(actual == expected)
+ }
+ }
+
+ "convert from VRL+RDW into VRL+RDW with RDW part of record length" in {
+ val expected = """{"T":"0"}{"T":"1"}{"T":"2"}{"T":"3"}"""
+ withTempDirectory("spark_cobol_processor") { tempDir =>
+ val binData = Array(
+ 0x00, 0x00, 0x05, 0x00, 0xF1,
+ 0x00, 0x00, 0x05, 0x00, 0xF2,
+ 0x00, 0x00, 0x05, 0x00, 0xF3,
+ 0x00, 0x00, 0x05, 0x00, 0xF4).map(_.toByte)
+
+ val inputPath = new Path(tempDir, "input.dat").toString
+ val outputPath = new Path(tempDir, "output").toString
+ val outputFile = new Path(outputPath, "input.dat").toString
+
+ writeBinaryFile(inputPath, binData)
+
+ SparkCobolProcessor.builder
+ .withCopybookContents(copybook)
+ .option("record_format", "V")
+ .option("is_rdw_part_of_record_length", "true")
+ .option("is_rdw_big_endian", "false")
+ .withProcessingStrategy(CobolProcessingStrategy.ToVariableLength)
+ .withRecordProcessor(new SerializableRawRecordProcessor {
+ override def processRecord(record: Array[Byte], ctx: CobolProcessorContext): Array[Byte] = {
+ record.map(v => (v - 1).toByte)
+ }
+ })
+ .load(inputPath)
+ .save(outputPath)
+
+ val outputData = readBinaryFile(outputFile)
+
+ assert(outputData.sameElements(
+ Array(0, 1, 0, 0, -16, 0, 1, 0, 0, -15, 0, 1, 0, 0, -14, 0, 1, 0, 0, -13).map(_.toByte)
+ ))
+
+ val actual = spark.read
+ .format("cobol")
+ .option("copybook_contents", copybook)
+ .option("record_format", "V")
+ .option("is_rdw_big_endian", "true")
+ .option("pedantic", "true")
+ .load(outputFile)
+ .toJSON
+ .collect()
+ .mkString
+
+ assert(actual == expected)
+ }
+ }
+
"support file_start_offset and file_end_offset with InPlace strategy" in {
val expected = """{"T":"0"}{"T":"1"}{"T":"2"}{"T":"3"}"""
withTempDirectory("spark_cobol_processor") { tempDir =>