#239 Fix row extraction for nested variable occurs case.

AbsaOSS · Feb 18, 2020 · 5fb28b8 · 5fb28b8
1 parent b2095a4
commit 5fb28b8
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 12 deletions.
diff --git a/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/RowExtractors.scala b/spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/RowExtractors.scala
@@ -80,8 +80,8 @@ object RowExtractors {
           var i = from
           var j = 0
           while (i < actualSize) {
-            val value = getGroupValues(offset, grp)
-            offset += grp.binaryProperties.dataSize
+            val (size, value) = getGroupValues(offset, grp)
+            offset += size
             groupValues(j) = value
             i += 1
             j += 1
@@ -107,11 +107,11 @@ object RowExtractors {
       }
     }
 
-    def extractValue(field: Statement, useOffset: Int): Any = {
+    def extractValue(field: Statement, useOffset: Int): (Int, Any) = {
       field match {
         case grp: Group =>
           if (grp.isSegmentRedefine && grp.name.compareToIgnoreCase(activeSegmentRedefine) != 0) {
-            null
+            (grp.binaryProperties.actualSize, null)
           } else {
             getGroupValues(useOffset, grp)
           }
@@ -125,11 +125,11 @@ object RowExtractors {
             }
             dependFields += st.name -> intVal
           }
-          value
+          (st.binaryProperties.actualSize, value)
       }
     }
 
-    def getGroupValues(offset: Int, group: Group): Row = {
+    def getGroupValues(offset: Int, group: Group): (Int, Row) = {
       var bitOffset = offset
 
       val fields = new Array[Any](group.nonFillerSize)
@@ -145,9 +145,13 @@ object RowExtractors {
           }
           value
         } else {
-          val value = extractValue(field, bitOffset)
+          val (size, value) = extractValue(field, bitOffset)
           if (!field.isRedefined) {
-            bitOffset += field.binaryProperties.actualSize
+            if (field.redefines.isDefined) {
+              bitOffset += field.binaryProperties.actualSize
+            } else {
+              bitOffset += size
+            }
           }
           value
         }
@@ -157,14 +161,14 @@ object RowExtractors {
         }
         i += 1
       }
-      new GenericRow(fields)
+      (bitOffset - offset, new GenericRow(fields))
     }
 
     var nextOffset = offsetBytes
 
     val records = for (record <- ast.children) yield {
-      val values = getGroupValues(nextOffset, record.asInstanceOf[Group])
-      nextOffset += record.binaryProperties.actualSize
+      val (size, values) = getGroupValues(nextOffset, record.asInstanceOf[Group])
+      nextOffset += size
       values
     }
 

diff --git a/...rc/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test21VariableOccurs.scala b/...rc/test/scala/za/co/absa/cobrix/spark/cobol/source/integration/Test21VariableOccurs.scala
@@ -29,7 +29,7 @@ import za.co.absa.cobrix.cobol.parser.policies.{CommentPolicy, StringTrimmingPol
 import za.co.absa.cobrix.cobol.parser.recordextractors.VarOccursRecordExtractor
 import za.co.absa.cobrix.cobol.parser.stream.{FSStream, SimpleStream}
 import za.co.absa.cobrix.spark.cobol.source.base.SparkTestBase
-import za.co.absa.cobrix.spark.cobol.utils.FileUtils
+import za.co.absa.cobrix.spark.cobol.utils.{FileUtils, RowExtractors}
 
 import scala.collection.immutable.HashMap
 import scala.collection.mutable.ListBuffer