#809 Make index cache 'true' by default, remove some code duplication.

Author: yruslan

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/VarLenNestedReader.scala

@@ -42,6 +42,8 @@ import scala.reflect.ClassTag
 class VarLenNestedReader[T: ClassTag](copybookContents: Seq[String],
                                       readerProperties: ReaderParameters,
                                       handler: RecordHandler[T]) extends VarLenReader with Logging with Serializable {
+  private val DEFAULT_INDEX_SIZE_COMPRESSED_FILES_MB = 1024
+  private val DEFAULT_FS_INDEX_SIZE_MULTIPLIER = 4
 
   protected val cobolSchema: CobolSchema = loadCopyBook(copybookContents)
 
@@ -217,13 +219,9 @@ class VarLenNestedReader[T: ClassTag](copybookContents: Seq[String],
 
   private def getSplitSizeMB(isCompressed: Boolean): Option[Int] = {
     if (isCompressed) {
-      readerProperties.inputSplitSizeCompressedMB.orElse(Some(1024))
+      readerProperties.inputSplitSizeCompressedMB.orElse(Some(DEFAULT_INDEX_SIZE_COMPRESSED_FILES_MB))
     } else {
-      if (readerProperties.inputSplitSizeMB.isDefined) {
-        readerProperties.inputSplitSizeMB
-      } else {
-        readerProperties.hdfsDefaultBlockSize
-      }
+      readerProperties.inputSplitSizeMB.orElse(readerProperties.fsDefaultBlockSize).map(_ * DEFAULT_FS_INDEX_SIZE_MULTIPLIER)
     }
   }
 

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

@@ -424,7 +424,7 @@ object CobolParametersParser extends Logging {
       inputSplitRecords = varLenParams.inputSplitRecords,
       inputSplitSizeMB = varLenParams.inputSplitSizeMB,
       inputSplitSizeCompressedMB = varLenParams.inputSplitSizeCompressedMB,
-      hdfsDefaultBlockSize = defaultBlockSize,
+      fsDefaultBlockSize = defaultBlockSize,
       startOffset = parameters.recordStartOffset,
       endOffset = parameters.recordEndOffset,
       fileStartOffset = varLenParams.fileStartOffset,
@@ -508,7 +508,7 @@ object CobolParametersParser extends Logging {
         fileEndOffset,
         isRecordIdGenerationEnabled,
         params.getOrElse(PARAM_ENABLE_INDEXES, "true").toBoolean,
-        params.getOrElse(PARAM_ENABLE_INDEX_CACHE, "false").toBoolean,
+        params.getOrElse(PARAM_ENABLE_INDEX_CACHE, "true").toBoolean,
         params.get(PARAM_INPUT_SPLIT_RECORDS).map(v => v.toInt),
         params.get(PARAM_INPUT_SPLIT_SIZE_MB).map(v => v.toInt),
         params.get(PARAM_INPUT_SPLIT_SIZE_COMPRESSED_MB).map(v => v.toInt),

cobol-parser/src/main/scala/za/co/absa/cobrix/cobol/reader/parameters/ReaderParameters.scala

@@ -51,7 +51,7 @@ import za.co.absa.cobrix.cobol.reader.policies.SchemaRetentionPolicy.SchemaReten
   * @param inputSplitRecords       The number of records to include in each partition. Notice mainframe records may have variable size, inputSplitMB is the recommended option
   * @param inputSplitSizeMB        A partition size to target. In certain circumstances this size may not be exactly that, but the library will do the best effort to target that size
   * @param inputSplitSizeCompressedMB A partition size to target for compressed files.
-  * @param hdfsDefaultBlockSize    Default HDFS block size for the HDFS filesystem used. This value is used as the default split size if inputSplitSizeMB is not specified
+  * @param fsDefaultBlockSize      Default HDFS block size for the HDFS filesystem used. This value is used as the default split size if inputSplitSizeMB is not specified
   * @param startOffset             An offset to the start of the record in each binary data block.
   * @param endOffset               An offset from the end of the record to the end of the binary data block.
   * @param fileStartOffset         A number of bytes to skip at the beginning of each file
@@ -104,7 +104,7 @@ case class ReaderParameters(
                              inputSplitRecords:       Option[Int] = None,
                              inputSplitSizeMB:        Option[Int] = None,
                              inputSplitSizeCompressedMB: Option[Int] = None,
-                             hdfsDefaultBlockSize:    Option[Int] = None,
+                             fsDefaultBlockSize:      Option[Int] = None,
                              startOffset:             Int = 0,
                              endOffset:               Int = 0,
                              fileStartOffset:         Int = 0,

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

@@ -174,7 +174,7 @@ object DefaultSource {
     */
   def createTextReader(parameters: CobolParameters, spark: SparkSession): FixedLenReader = {
     val copybookContent = CopybookContentLoader.load(parameters, spark.sparkContext.hadoopConfiguration)
-    val defaultHdfsBlockSize = SparkUtils.getDefaultHdfsBlockSize(spark, parameters.sourcePaths.headOption)
+    val defaultHdfsBlockSize = SparkUtils.getDefaultFsBlockSize(spark, parameters.sourcePaths.headOption)
     new FixedLenTextReader(copybookContent,  getReaderProperties(parameters, defaultHdfsBlockSize)
     )
   }
@@ -185,7 +185,7 @@ object DefaultSource {
   def createFixedLengthReader(parameters: CobolParameters, spark: SparkSession): FixedLenReader = {
 
     val copybookContent = CopybookContentLoader.load(parameters, spark.sparkContext.hadoopConfiguration)
-    val defaultHdfsBlockSize = SparkUtils.getDefaultHdfsBlockSize(spark, parameters.sourcePaths.headOption)
+    val defaultHdfsBlockSize = SparkUtils.getDefaultFsBlockSize(spark, parameters.sourcePaths.headOption)
     new FixedLenNestedReader(copybookContent, getReaderProperties(parameters, defaultHdfsBlockSize)
     )
   }
@@ -199,7 +199,7 @@ object DefaultSource {
 
 
     val copybookContent = CopybookContentLoader.load(parameters, spark.sparkContext.hadoopConfiguration)
-    val defaultHdfsBlockSize = SparkUtils.getDefaultHdfsBlockSize(spark, parameters.sourcePaths.headOption)
+    val defaultHdfsBlockSize = SparkUtils.getDefaultFsBlockSize(spark, parameters.sourcePaths.headOption)
     new VarLenNestedReader(
       copybookContent, getReaderProperties(parameters, defaultHdfsBlockSize)
     )

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

@@ -18,7 +18,7 @@ package za.co.absa.cobrix.spark.cobol.source.streaming
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FSDataInputStream, Path}
-import org.apache.hadoop.io.compress.CompressionCodecFactory
+import za.co.absa.cobrix.spark.cobol.utils.FileUtils
 
 import java.io.{IOException, InputStream}
 
@@ -121,11 +121,9 @@ class BufferedFSDataInputStream(filePath: Path, hadoopConfig: Configuration, sta
 
   private def openStream(): InputStream = {
     val fileSystem = filePath.getFileSystem(hadoopConfig)
+    val codec = FileUtils.getCompressionCodec(filePath, hadoopConfig)
     val fsIn: FSDataInputStream = fileSystem.open(filePath)
 
-    val factory = new CompressionCodecFactory(hadoopConfig)
-    val codec = factory.getCodec(filePath)
-
     val baseStream = if (codec != null) {
       isCompressedStream = true
       codec.createInputStream(fsIn)

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

@@ -18,10 +18,10 @@ package za.co.absa.cobrix.spark.cobol.source.streaming
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{ContentSummary, Path}
-import org.apache.hadoop.io.compress.CompressionCodecFactory
 import org.apache.log4j.Logger
 import za.co.absa.cobrix.cobol.reader.common.Constants
 import za.co.absa.cobrix.cobol.reader.stream.SimpleStream
+import za.co.absa.cobrix.spark.cobol.utils.FileUtils
 
 import java.io.IOException
 
@@ -51,12 +51,7 @@ class FileStreamer(filePath: String, hadoopConfig: Configuration, startOffset: L
   private var wasOpened = false
   private var bufferedStream: BufferedFSDataInputStream = _
 
-  private lazy val isCompressedStream = {
-    val factory = new CompressionCodecFactory(hadoopConfig)
-    val codec = factory.getCodec(hadoopPath)
-
-    codec != null
-  }
+  private lazy val isCompressedStream = FileUtils.isCompressed(hadoopPath, hadoopConfig)
 
   private lazy val fileSize = getHadoopFileSize(hadoopPath)
 

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/FileUtils.scala

@@ -18,7 +18,7 @@ package za.co.absa.cobrix.spark.cobol.utils
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs._
-import org.apache.hadoop.io.compress.CompressionCodecFactory
+import org.apache.hadoop.io.compress.{CompressionCodec, CompressionCodecFactory}
 import za.co.absa.cobrix.cobol.internal.Logging
 
 import java.io.{FileOutputStream, IOException, OutputStreamWriter, PrintWriter}
@@ -217,16 +217,17 @@ object FileUtils extends Logging {
   }
 
   def isCompressed(file: Path, hadoopConfig: Configuration): Boolean = {
-    val factory = new CompressionCodecFactory(hadoopConfig)
-    val codec = factory.getCodec(file)
+    getCompressionCodec(file, hadoopConfig) != null
+  }
 
-    codec != null
+  def getCompressionCodec(file: Path, hadoopConfig: Configuration): CompressionCodec = {
+    val factory = new CompressionCodecFactory(hadoopConfig)
+    factory.getCodec(file)
   }
 
   def getCompressedFileSize(file: Path, hadoopConfig: Configuration): Long = {
     logger.warn(s"Using full scan to determine file size of $file..")
-    val factory = new CompressionCodecFactory(hadoopConfig)
-    val codec = factory.getCodec(file)
+    val codec = getCompressionCodec(file, hadoopConfig)
     val fileSystem = file.getFileSystem(hadoopConfig)
     val fsIn: FSDataInputStream = fileSystem.open(file)
     val ifs = codec.createInputStream(fsIn)

spark-cobol/src/main/scala/za/co/absa/cobrix/spark/cobol/utils/SparkUtils.scala

@@ -19,12 +19,12 @@ package za.co.absa.cobrix.spark.cobol.utils
 import com.fasterxml.jackson.databind.ObjectMapper
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.SparkContext
-import org.apache.spark.sql.functions.{array, col, expr, max, struct}
-import za.co.absa.cobrix.spark.cobol.utils.impl.HofsWrapper.transform
+import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, DataFrame, SparkSession}
 import za.co.absa.cobrix.cobol.internal.Logging
 import za.co.absa.cobrix.spark.cobol.parameters.MetadataFields.MAX_ELEMENTS
+import za.co.absa.cobrix.spark.cobol.utils.impl.HofsWrapper.transform
 
 import scala.annotation.tailrec
 import scala.collection.mutable
@@ -489,7 +489,7 @@ object SparkUtils extends Logging {
     }.getOrElse(None)
   }
 
-  def getDefaultHdfsBlockSize(spark: SparkSession, pathOpt: Option[String]): Option[Int] = {
+  def getDefaultFsBlockSize(spark: SparkSession, pathOpt: Option[String]): Option[Int] = {
     val conf = spark.sparkContext.hadoopConfiguration
 
     val fileSystem  =pathOpt match {

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

@@ -56,6 +56,7 @@ class Test40CompressesFilesSpec extends AnyFunSuite with SparkTestBase with Bina
     val options = if (useIndexes) {
       Map(
         "input_split_records" -> "1",
+        "enable_index_cache" -> "false",
         "generate_record_id" -> "true"
       )
     } else {
@@ -178,6 +179,8 @@ class Test40CompressesFilesSpec extends AnyFunSuite with SparkTestBase with Bina
       .option("schema_retention_policy", "collapse_root")
       .option("floating_point_format", "IEEE754")
       .option("strict_sign_overpunching", "true")
+      .option("generate_record_id", "true")
+      .option("enable_index_cache", "false")
       .option("pedantic", "true")
       .load(inputDataPath)
 
@@ -195,6 +198,7 @@ class Test40CompressesFilesSpec extends AnyFunSuite with SparkTestBase with Bina
       .option("floating_point_format", "IEEE754")
       .option("strict_sign_overpunching", "true")
       .option("file_end_offset", 1493)
+      .option("enable_index_cache", "false")
       .option("pedantic", "true")
       .load(inputDataPath)