#809 Fix file end offset setting for compressed files.

Author: yruslan

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/scanners/CobolScanners.scala

@@ -18,6 +18,7 @@ package za.co.absa.cobrix.spark.cobol.source.scanners
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
+import org.apache.hadoop.io.compress.CompressionCodecFactory
 import org.apache.hadoop.io.{LongWritable, Text}
 import org.apache.hadoop.mapred.TextInputFormat
 import org.apache.spark.rdd.RDD
@@ -72,7 +73,17 @@ private[source] object CobolScanners extends Logging {
           val maximumFileBytes = if (reader.getReaderProperties.fileEndOffset == 0) {
             0
           } else {
-            fileSystem.getFileStatus(path).getLen - reader.getReaderProperties.fileEndOffset - startFileOffset
+            if (isCompressed(path, sconf.value)) {
+              // ToDo determine if the uncompressed file size can be effectively fetched
+              if (reader.getReaderProperties.fileEndOffset > 0) {
+                logger.warn(s"File end offset for $path is ignored because the file is compressed.")
+              }
+              0L
+            } else {
+              val fileSize = fileSystem.getFileStatus(path).getLen
+
+              fileSize - reader.getReaderProperties.fileEndOffset - startFileOffset
+            }
           }
           val dataStream = new FileStreamer(filePath, sconf.value, startFileOffset, maximumFileBytes)
           val headerStream = new FileStreamer(filePath, sconf.value, startFileOffset)
@@ -81,6 +92,13 @@ private[source] object CobolScanners extends Logging {
       })
   }
 
+  private[source] def isCompressed(file: Path, hadoopConfig: Configuration): Boolean = {
+    val factory = new CompressionCodecFactory(hadoopConfig)
+    val codec = factory.getCodec(file)
+
+    codec != null
+  }
+
   private[source] def buildScanForFixedLength(reader: FixedLenReader, sourceDirs: Seq[String],
                                               recordParser: (FixedLenReader, RDD[Array[Byte]]) => RDD[Row],
                                               debugIgnoreFileSize: Boolean,

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/source/streaming/BufferedFSDataInputStream.scala

@@ -86,8 +86,10 @@ class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, sta
           offsetLeft += lengthLeft
           lengthLeft = 0
         } else {
-          if (bufferContainBytes > 0) {
-            System.arraycopy(buffer, bufferPos, b, offsetLeft, lengthLeft)
+          if (bufferContainBytes > 0 && lengthLeft > 0) {
+            val available = bufferContainBytes - bufferPos
+            val bytesToCopy = Math.min(lengthLeft, available)
+            System.arraycopy(buffer, bufferPos, b, offsetLeft, bytesToCopy)
             bufferPos += bufferContainBytes
             offsetLeft += bufferContainBytes
             lengthLeft -= bufferContainBytes
@@ -128,12 +130,26 @@ class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, sta
     val factory = new CompressionCodecFactory(hadoopConfig)
     val codec = factory.getCodec(filePath)
 
-    if (codec != null) {
+    val baseStream = if (codec != null) {
       isCompressedStream = true
       codec.createInputStream(fsIn)
     } else {
       // No compression detected
       fsIn
     }
+
+    if (startOffset > 0) {
+      if (codec == null) {
+        baseStream.seek(startOffset)
+      } else {
+        var toSkip = startOffset
+        while (toSkip > 0) {
+          val skipped = baseStream.skip(toSkip)
+          if (skipped <= 0) return baseStream
+          toSkip -= skipped
+        }
+      }
+    }
+    baseStream
   }
 }