From 329b2a51c0d535b6530b2932fa61f5d1c6b3326c Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Sat, 29 Aug 2026 07:23:20 +0200 Subject: [PATCH 1/2] #870 Add support for VRL files to be processed via in-place SparkCobolProcessor. --- .../headerparsers/RecordHeaderParserRDW.scala | 18 +-- .../cobol/reader/VarLenNestedReader.scala | 2 + .../FixedRecordLengthRawRecordExtractor.scala | 43 +++++-- ...thRecordLengthExprRawRecordExtractor.scala | 25 ++++ .../extractors/raw/RawRecordContext.scala | 8 +- ...VariableBlockVariableRecordExtractor.scala | 70 ++++++++--- .../VariableRecordLengthRecordExtractor.scala | 115 ++++++++++++++++++ .../recordheader/RecordHeaderDecoderRdw.scala | 2 +- .../recordheader/RecordHeaderParameters.scala | 4 +- .../impl/CobolProcessorBaseSuite.scala | 38 +++--- .../RecordHeaderParametersFactory.scala | 5 +- .../cobol/SparkCobolProcessorSuite.scala | 100 +++++++++++++++ 12 files changed, 371 insertions(+), 59 deletions(-) create mode 100644 cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/VariableRecordLengthRecordExtractor.scala 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/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 => From 35fbe7ddd4debdbc708380359bac7dd2f3f67ba7 Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Sun, 30 Aug 2026 14:39:04 +0200 Subject: [PATCH 2/2] Bump ScalaTest to 3.2.20 and fix GPG decryption example in changelog. --- README.md | 6 +++--- pom.xml | 2 +- project/Dependencies.scala | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) 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/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"