#420 Add support for FB format with BDW only.

Author: yruslan

README.md

@@ -326,8 +326,21 @@ the record defined by the copybook. But you can specify the record length explic
 ```
 
 Fixed block record formats (`FB`) are also supported. The support is _experimental_, if you find any issues, please
-let us know. When the record format is 'FB' you need to specify either block length or number of records per
+let us know. When the record format is 'FB' you can specify block length or number of records per
 block. As with 'F' if `record_length` is not specified, it will be determined from the copybook.
+
+Records that have BDWs, but not rdws can be read like this:
+```
+.option("record_format", "FB")
+.option("record_length", "250")
+```
+or simply
+```
+.option("record_format", "FB")
+```
+
+Records that have neither BDWs nor RDWs can be read like this:
+
 ```
 .option("record_format", "FB")
 .option("record_length", "250")

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

@@ -24,9 +24,6 @@ case class FixedBlockParameters(
 
 object FixedBlockParameters {
   def validate(params: FixedBlockParameters): Unit = {
-    if (params.blockLength.isEmpty && params.recordsPerBlock.isEmpty) {
-      throw new IllegalArgumentException("FB record format requires block length or number records per block to be specified.")
-    }
     if (params.blockLength.nonEmpty && params.recordsPerBlock.nonEmpty) {
       throw new IllegalArgumentException("FB record format requires either block length or number records per block to be specified, but not both.")
     }

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

@@ -22,7 +22,7 @@ class FixedBlockRawRecordExtractor(ctx: RawRecordContext, fbParams: FixedBlockPa
   private val recordQueue = new mutable.Queue[Array[Byte]]
 
   private val recordSize = fbParams.recordLength.getOrElse(ctx.copybook.getRecordSize)
-  private val bdwSize = fbParams.blockLength.getOrElse(fbParams.recordsPerBlock.get * recordSize)
+  private val bdwSize = fbParams.blockLength.orElse(fbParams.recordsPerBlock.map(_ * recordSize))
 
   override def offset: Long = ctx.inputStream.offset
 
@@ -35,14 +35,20 @@ class FixedBlockRawRecordExtractor(ctx: RawRecordContext, fbParams: FixedBlockPa
 
   private def readNextBlock(): Unit = {
     if (!ctx.inputStream.isEndOfStream) {
-      val bdwOffset = ctx.inputStream.offset
-      val blockBuffer = ctx.inputStream.next(bdwSize)
+      var bdwOffset = ctx.inputStream.offset
+
+      val nextBlockSize = bdwSize.getOrElse({
+        val bdw = ctx.inputStream.next(ctx.bdwDecoder.headerSize)
+        val blockLength = ctx.bdwDecoder.getRecordLength(bdw, bdwOffset)
+        bdwOffset += ctx.bdwDecoder.headerSize
+        blockLength
+      })
+
+      val blockBuffer = ctx.inputStream.next(nextBlockSize)
 
       var blockIndex = 0
 
       while (blockIndex < blockBuffer.length) {
-        val rdwOffset = bdwOffset + blockIndex
-
         val payload = blockBuffer.slice(blockIndex, blockIndex + recordSize)
         if (payload.length > 0) {
           recordQueue.enqueue(payload)

cobol-parser/src/test/scala/za/co/absa/cobrix/cobol/reader/extractors/raw/FixedBlockRawRecordExtractorSuite.scala

@@ -99,19 +99,16 @@ class FixedBlockRawRecordExtractorSuite extends WordSpec {
       assert(extractor.next().head == 0xF8.toByte)
       assert(!extractor.hasNext)
     }
-  }
 
-  "failures" should {
-    "throw an exception when neither block length nor records per block is specified" in {
+    "allow neither block length nor records per block to be specified" in {
       val fb = FixedBlockParameters(Some(1), None, None)
 
-      val ex = intercept[IllegalArgumentException] {
-        FixedBlockParameters.validate(fb)
-      }
-
-      assert(ex.getMessage.contains("FB record format requires block length or number records per block to be specified."))
+      assert(fb.blockLength.isEmpty)
+      assert(fb.recordsPerBlock.isEmpty)
     }
+  }
 
+  "failures" should {
     "throw an exception when both block length and records per block are specified" in {
       val fb = FixedBlockParameters(Some(1), Some(1), Some(1))
 

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/parameters/CobolParametersParser.scala

@@ -325,9 +325,6 @@ object CobolParametersParser {
       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 == FixedBlock && bdw.blockLength.isEmpty && bdw.recordsPerBlock.isEmpty ) {
-        throw new IllegalArgumentException(s"For FB file format either '$PARAM_BLOCK_LENGTH' or '$PARAM_RECORDS_PER_BLOCK' must be specified.")
-      }
       if (recordFormat == VariableBlock && bdw.blockLength.nonEmpty) {
         logger.warn(s"Option '$PARAM_BLOCK_LENGTH' is ignored for record format: VB")
       }

spark-cobol/src/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test30FbFileSpec.scala

@@ -54,6 +54,20 @@ class Test30FbFileSpec extends WordSpec with SparkTestBase with BinaryFileFixtur
         """[{"A":"0"},{"A":"0"},{"A":"1"},{"A":"1"}]""",
         ignoreCount = true)
     }
+
+    "load data with BDWs" in {
+      testVbRecordLoad(2, 2,
+        Map[String, String](),
+        expected4Records,
+        hasBDW = true)
+    }
+
+    "load empty data with BDWs" in {
+      testVbRecordLoad(0, 0,
+        Map[String, String](),
+        "[]",
+        hasBDW = true)
+    }
   }
 
   "FB record failures should happen" when {
@@ -80,31 +94,31 @@ class Test30FbFileSpec extends WordSpec with SparkTestBase with BinaryFileFixtur
 
       assert(ex.getMessage.contains("Options 'block_length' and 'records_per_block' cannot be used together."))
     }
-
-    "Mandatory options are missing" in {
-      val ex = intercept[IllegalArgumentException] {
-        testVbRecordLoad(1, 2, Map[String, String](), "")
-      }
-
-      assert(ex.getMessage.contains("For FB file format either 'block_length' or 'records_per_block' must be specified."))
-    }
   }
 
   private def testVbRecordLoad(blocks: Int,
                                records: Int,
                                options: Map[String, String],
                                expected: String,
-                               ignoreCount: Boolean = false): Unit = {
+                               ignoreCount: Boolean = false,
+                               hasBDW: Boolean = false): Unit = {
     val record: Seq[Byte] = Range(0, blocks).flatMap(blockNum => {
-      Range(0, records).flatMap(recordNum => {
+      val bdw: Seq[Byte] = if (hasBDW) {
+        val byte0 = ((records * 2) % 256).toByte
+        val byte1 = ((records * 2) / 256).toByte
+        Seq(0, 0, byte0, byte1)
+      } else {
+        Nil
+      }
+      bdw ++ Range(0, records).flatMap(recordNum => {
         val idx = (blockNum * records + recordNum) % 10
         val v = (0xF0 + idx).toByte
         Seq(v, v)
       })
     })
 
     withTempBinFile("rec", ".dat", record.toArray) { tmpFileName1 =>
-      val df =     spark
+      val df = spark
         .read
         .format("cobol")
         .option("copybook_contents", copybook)