diff --git a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala index d48332a4..70535602 100644 --- a/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala +++ b/cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/CobolParametersParser.scala @@ -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.") @@ -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 @@ -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, @@ -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, @@ -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") } if (recordFormat == FixedBlock && bdw.recordsPerBlock.nonEmpty) { diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala index ab377627..7d5fbe79 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidator.scala @@ -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" + } + } + if (readerParameters.occursMappings.nonEmpty) { issues += "OCCURS mapping option ('occurs_mappings') is not supported for writing" } diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer/NestedRecordCombiner.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer/NestedRecordCombiner.scala index 4212752b..9e77fabf 100644 --- a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer/NestedRecordCombiner.scala +++ b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/writer/NestedRecordCombiner.scala @@ -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 @@ -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, @@ -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. * diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidatorSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidatorSuite.scala index 54ae9a2f..0ca38e72 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidatorSuite.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/parameters/CobolParametersValidatorSuite.scala @@ -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")) diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableBlockEbcdicWriterSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableBlockEbcdicWriterSuite.scala new file mode 100644 index 00000000..f171f0af --- /dev/null +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableBlockEbcdicWriterSuite.scala @@ -0,0 +1,289 @@ +/* + * 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.spark.cobol.writer + +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.SaveMode +import org.scalatest.Assertion +import org.scalatest.wordspec.AnyWordSpec +import za.co.absa.cobrix.spark.cobol.source.base.SparkTestBase +import za.co.absa.cobrix.spark.cobol.source.fixtures.{BinaryFileFixture, TextComparisonFixture} + +/** + * Tests for writing VB (Variable Block) COBOL files. + * + * A VB file groups one or more variable-length records into "blocks". On disk the layout is: + * + * BDW (4 bytes) | RDW (4 bytes) + payload | RDW (4 bytes) + payload | ... <- block 1 + * BDW (4 bytes) | RDW (4 bytes) + payload | ... <- block 2 + * + * where: + * - RDW (Record Descriptor Word) prefixes each record and encodes the payload length. + * - BDW (Block Descriptor Word) prefixes each block and encodes the total length of everything + * that follows it in the block (the sum of all RDW+payload bytes), EXCLUDING the BDW's own 4 bytes. + */ +class VariableBlockEbcdicWriterSuite extends AnyWordSpec with SparkTestBase with BinaryFileFixture with TextComparisonFixture { + + import spark.implicits._ + + // A = PIC X(1), B = PIC X(5) => payload is 6 bytes per record. + // With a 4-byte RDW prefix, each record occupies 10 bytes on disk. + private val copybookContents = + """ 01 RECORD. + 05 A PIC X(1). + 05 B PIC X(5). + """ + + // Reusable per-record byte fragments (RDW + EBCDIC payload) for the sample rows below. + // RDW little-endian, payload length 6 => 0x06 0x00 0x00 0x00 + private val rdwLe = Array[Byte](0x06, 0x00, 0x00, 0x00) + // RDW big-endian, payload length 6 => 0x00 0x06 0x00 0x00 + private val rdwBe = Array[Byte](0x00, 0x06, 0x00, 0x00) + + private val payloadA = Array[Byte](0xC1.toByte, 0xC6.toByte, 0x89.toByte, 0x99.toByte, 0xA2.toByte, 0xA3.toByte) // "A","First" + private val payloadB = Array[Byte](0xC2.toByte, 0xE2.toByte, 0x83.toByte, 0x95.toByte, 0x84.toByte, 0x40.toByte) // "B","Scnd_" + private val payloadC = Array[Byte](0xC3.toByte, 0xD3.toByte, 0x81.toByte, 0xA2.toByte, 0xA3.toByte, 0x40.toByte) // "C","Last_" + private val payloadD = Array[Byte](0xC4.toByte, 0xC6.toByte, 0x96.toByte, 0x99.toByte, 0xA3.toByte, 0x88.toByte) // "D","Forth" + + private val fourRows = List(("A", "First"), ("B", "Scnd"), ("C", "Last"), ("D", "Forth")) + private val threeRows = List(("A", "First"), ("B", "Scnd"), ("C", "Last")) + private val twoRows = List(("A", "First"), ("B", "Scnd")) + + // BDW encoders. blockLength is the sum of the (RDW + payload) bytes of every record in the block. + private def bdwLe(blockLength: Int): Array[Byte] = + Array[Byte](0x00, 0x00, (blockLength & 0xFF).toByte, ((blockLength >> 8) & 0xFF).toByte) + + private def bdwBe(blockLength: Int): Array[Byte] = + Array[Byte](((blockLength >> 8) & 0xFF).toByte, (blockLength & 0xFF).toByte, 0x00, 0x00) + + "cobol VB writer" should { + "write a VB file with records_per_block = 2, little-endian BDW and RDW" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = fourRows.toDF("A", "B") + val path = new Path(tempDir, "vb1") + + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("records_per_block", "2") + .save(path.toString) + + // Two blocks of two records each. Each block payload = 2 * (4 + 6) = 20 bytes. + val expected = + bdwLe(20) ++ rdwLe ++ payloadA ++ rdwLe ++ payloadB ++ + bdwLe(20) ++ rdwLe ++ payloadC ++ rdwLe ++ payloadD + + assertArraysEqual(readSinglePartFile(path), expected) + } + } + + "write a VB file with records_per_block = 2, big-endian BDW and RDW" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = fourRows.toDF("A", "B") + val path = new Path(tempDir, "vb2") + + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("records_per_block", "2") + .option("is_bdw_big_endian", "true") + .option("is_rdw_big_endian", "true") + .save(path.toString) + + val expected = + bdwBe(20) ++ rdwBe ++ payloadA ++ rdwBe ++ payloadB ++ + bdwBe(20) ++ rdwBe ++ payloadC ++ rdwBe ++ payloadD + + assertArraysEqual(readSinglePartFile(path), expected) + } + } + + "write a VB file with records_per_block = 2 and an uneven number of records" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = threeRows.toDF("A", "B") + val path = new Path(tempDir, "vb3") + + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("records_per_block", "2") + .save(path.toString) + + // First block has 2 records (blockLength 20), last block has the leftover single record (blockLength 10). + val expected = + bdwLe(20) ++ rdwLe ++ payloadA ++ rdwLe ++ payloadB ++ + bdwLe(10) ++ rdwLe ++ payloadC + + assertArraysEqual(readSinglePartFile(path), expected) + } + } + + "write a VB file with block_length cap producing an uneven split" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = threeRows.toDF("A", "B") + val path = new Path(tempDir, "vb4") + + // Each record is 10 bytes. With a 25-byte cap, 2 records (20 bytes) fit, a 3rd (30) would overflow. + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("block_length", "25") + .save(path.toString) + + val expected = + bdwLe(20) ++ rdwLe ++ payloadA ++ rdwLe ++ payloadB ++ + bdwLe(10) ++ rdwLe ++ payloadC + + assertArraysEqual(readSinglePartFile(path), expected) + } + } + + "write a VB file where a single record exceeds block_length (record is never split)" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = twoRows.toDF("A", "B") + val path = new Path(tempDir, "vb5") + + // block_length smaller than a single 10-byte record => each record gets its own block. + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("block_length", "5") + .save(path.toString) + + val expected = + bdwLe(10) ++ rdwLe ++ payloadA ++ + bdwLe(10) ++ rdwLe ++ payloadB + + assertArraysEqual(readSinglePartFile(path), expected) + } + } + + "fail fast when writing VB without a blocking option" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = twoRows.toDF("A", "B") + val path = new Path(tempDir, "vb6") + + val exception = intercept[IllegalArgumentException] { + df.coalesce(1) + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .save(path.toString) + } + + assert(exception.getMessage.contains("records_per_block")) + assert(exception.getMessage.contains("block_length")) + } + } + + "fail when writing VB with both blocking options" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = twoRows.toDF("A", "B") + val path = new Path(tempDir, "vb7") + + val exception = intercept[IllegalArgumentException] { + df.coalesce(1) + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("records_per_block", "2") + .option("block_length", "25") + .save(path.toString) + } + + assert(exception.getMessage.contains("cannot be used together")) + } + } + + "round-trip: write a VB file and read it back" in { + withTempDirectory("cobol_vb_writer") { tempDir => + val df = fourRows.toDF("A", "B") + val path = new Path(tempDir, "vb8") + + df.coalesce(1) + .orderBy("A") + .write + .format("cobol") + .mode(SaveMode.Overwrite) + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .option("records_per_block", "2") + .save(path.toString) + + val readBack = spark.read + .format("cobol") + .option("copybook_contents", copybookContents) + .option("record_format", "VB") + .load(path.toString) + .orderBy("A") + .collect() + .map(r => (r.getString(0), r.getString(1))) + .toList + + assert(readBack == List(("A", "First"), ("B", "Scnd"), ("C", "Last"), ("D", "Forth"))) + } + } + } + + private def readSinglePartFile(path: Path): Array[Byte] = { + val fs = path.getFileSystem(spark.sparkContext.hadoopConfiguration) + assert(fs.exists(path), "Output directory should exist") + val files = fs.listStatus(path).filter(_.getPath.getName.startsWith("part-")) + assert(files.nonEmpty, "Output directory should contain part files") + + val partFile = files.head.getPath + val data = fs.open(partFile) + val bytes = new Array[Byte](files.head.getLen.toInt) + data.readFully(bytes) + data.close() + bytes + } + + private def assertArraysEqual(actual: Array[Byte], expected: Array[Byte]): Assertion = { + if (!actual.sameElements(expected)) { + val actualHex = actual.map(b => f"0x$b%02X").mkString(", ") + val expectedHex = expected.map(b => f"0x$b%02X").mkString(", ") + fail(s"Actual: $actualHex\nExpected: $expectedHex") + } else { + succeed + } + } +} diff --git a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableLengthEbcdicWriterSuite.scala b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableLengthEbcdicWriterSuite.scala index 9544db3f..6c1be4dc 100644 --- a/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableLengthEbcdicWriterSuite.scala +++ b/spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/writer/VariableLengthEbcdicWriterSuite.scala @@ -153,7 +153,7 @@ class VariableLengthEbcdicWriterSuite extends AnyWordSpec with SparkTestBase wit .save(path.toString) } - assert(exception.getMessage.contains("Writer validation issues: Only 'F' and 'V' values for 'record_format' are supported for writing, provided value: 'FB';")) + assert(exception.getMessage.contains("Writer validation issues: Only 'F', 'V' and 'VB' values for 'record_format' are supported for writing, provided value: 'FB';")) assert(exception.getMessage.contains("OCCURS mapping option ('occurs_mappings') is not supported for writing")) assert(exception.getMessage.contains("'record_start_offset' and 'record_end_offset' are not supported for writing")) assert(exception.getMessage.contains("'file_start_offset' and 'file_end_offset' are not supported for writing"))