#805 Fix PR suggestions (Thanks @codarabbitai).

Author: yruslan

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

@@ -832,6 +832,10 @@ object CobolParametersParser extends Logging {
       }
     }
 
+    if (!params.contains(PARAM_RECORD_LENGTH_FIELD) && params.contains(PARAM_RECORD_LENGTH_MAP)) {
+      throw new IllegalArgumentException(s"Option '$PARAM_RECORD_LENGTH_MAP' requires '$PARAM_RECORD_LENGTH_FIELD' to be specified.")
+    }
+
     if (params.contains(PARAM_RECORD_LENGTH)) {
       val incorrectParameters = new ListBuffer[String]
       if (isText) {
@@ -950,6 +954,10 @@ object CobolParametersParser extends Logging {
       params.contains(PARAM_STRICT_INTEGRAL_PRECISION) && params(PARAM_STRICT_INTEGRAL_PRECISION).toBoolean)
       throw new IllegalArgumentException(s"Options '$PARAM_DISPLAY_PIC_ALWAYS_STRING' and '$PARAM_STRICT_INTEGRAL_PRECISION' cannot be used together.")
 
+    if (params.contains(PARAM_ENABLE_INDEXES) && !params(PARAM_ENABLE_INDEXES).toBoolean &&
+      params.contains(PARAM_ENABLE_INDEX_CACHE) && params(PARAM_ENABLE_INDEX_CACHE).toBoolean)
+      throw new IllegalArgumentException(s"When '$PARAM_ENABLE_INDEXES' = false, '$PARAM_ENABLE_INDEX_CACHE' cannot be true.")
+
     if (validateRedundantOptions && unusedKeys.nonEmpty) {
       val unusedKeyStr = unusedKeys.mkString(",")
       val msg = s"Redundant or unrecognized option(s) to 'spark-cobol': $unusedKeyStr."

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

@@ -65,16 +65,15 @@ class SerializableConfiguration(@transient var value: Configuration) extends Ser
 class CobolRelation(sourceDirs: Seq[String],
                     cobolReader: Reader,
                     localityParams: LocalityParameters,
-                    debugIgnoreFileSize: Boolean,
-                    indexCachingAllowed: Boolean)
+                    debugIgnoreFileSize: Boolean)
                    (@transient val sqlContext: SQLContext)
   extends BaseRelation
     with Serializable
     with TableScan {
 
   private val filesList = CobolRelation.getListFilesWithOrder(sourceDirs, sqlContext, isRecursiveRetrieval)
 
-  private lazy val indexes: RDD[SparseIndexEntry] = IndexBuilder.buildIndex(filesList, cobolReader, sqlContext, indexCachingAllowed)(localityParams)
+  private lazy val indexes: RDD[SparseIndexEntry] = IndexBuilder.buildIndex(filesList, cobolReader, sqlContext, cobolReader.getReaderProperties.isIndexCachingAllowed)(localityParams)
 
   override def schema: StructType = {
     cobolReader.getSparkSchema

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

@@ -58,16 +58,11 @@ class DefaultSource
 
     val cobolParameters = CobolParametersParser.parse(new Parameters(parameters))
     CobolParametersValidator.checkSanity(cobolParameters)
-    val indexCachingAllowed = cobolParameters.variableLengthParams match {
-      case Some(varLenParams) => varLenParams.isIndexCachingAllowed
-      case None => false
-    }
 
     new CobolRelation(cobolParameters.sourcePaths,
       buildEitherReader(sqlContext.sparkSession, cobolParameters),
       LocalityParameters.extract(cobolParameters),
-      cobolParameters.debugIgnoreFileSize,
-      indexCachingAllowed)(sqlContext)
+      cobolParameters.debugIgnoreFileSize)(sqlContext)
   }
 
   /** Writer relation */

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

@@ -46,7 +46,7 @@ import scala.collection.mutable.ArrayBuffer
   * In a nutshell, ideally, there will be as many partitions as are there are indexes.
   */
 private[cobol] object IndexBuilder extends Logging {
-  private val indexCache = new ConcurrentHashMap[String, Array[SparseIndexEntry]]()
+  private[cobol] val indexCache = new ConcurrentHashMap[String, Array[SparseIndexEntry]]()
 
   def buildIndex(filesList: Array[FileWithOrder],
                  cobolReader: Reader,
@@ -124,7 +124,7 @@ private[cobol] object IndexBuilder extends Logging {
     val conf = sqlContext.sparkContext.hadoopConfiguration
     val sconf = new SerializableConfiguration(conf)
 
-    // Splitting between files for which indexes are chached and teh list of files for which indexes are not cached
+    // Splitting between files for which indexes are cached and the list of files for which indexes are not cached
     val cachedFiles = if (cachingAllowed) {
       filesList.filter(f => indexCache.containsKey(f.filePath))
     } else {
@@ -381,7 +381,7 @@ private[cobol] object IndexBuilder extends Logging {
   private def createIndexRDD(indexes: Array[SparseIndexEntry], sqlContext: SQLContext): RDD[SparseIndexEntry] = {
     val indexCount = indexes.length
 
-    val numPartitions = Math.min(indexCount, Constants.maxNumPartitions)
+    val numPartitions = Math.max(1, Math.min(indexCount, Constants.maxNumPartitions))
     logger.info(s"Index elements count: ${indexes.length}, number of partitions = $numPartitions")
 
     sqlContext.sparkContext.parallelize(indexes, numPartitions)

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

@@ -64,8 +64,7 @@ class CobolRelationSpec extends SparkCobolTestBase with Serializable {
     val relation = new CobolRelation(Seq(copybookFile.getParentFile.getAbsolutePath),
       testReader,
       localityParams = localityParams,
-      debugIgnoreFileSize = false,
-      indexCachingAllowed = false)(sqlContext)
+      debugIgnoreFileSize = false)(sqlContext)
     val cobolData: RDD[Row] = relation.parseRecords(testReader, oneRowRDD)
 
     val cobolDataFrame = sqlContext.createDataFrame(cobolData, sparkSchema)
@@ -89,8 +88,7 @@ class CobolRelationSpec extends SparkCobolTestBase with Serializable {
     val relation = new CobolRelation(Seq(copybookFile.getParentFile.getAbsolutePath),
       testReader,
       localityParams = localityParams,
-      debugIgnoreFileSize = false,
-      indexCachingAllowed = false)(sqlContext)
+      debugIgnoreFileSize = false)(sqlContext)
 
     val caught = intercept[Exception] {
       relation.parseRecords(testReader, oneRowRDD).collect()
@@ -105,8 +103,7 @@ class CobolRelationSpec extends SparkCobolTestBase with Serializable {
     val relation = new CobolRelation(Seq(copybookFile.getParentFile.getAbsolutePath),
       testReader,
       localityParams = localityParams,
-      debugIgnoreFileSize = false,
-      indexCachingAllowed = false)(sqlContext)
+      debugIgnoreFileSize = false)(sqlContext)
 
     val caught = intercept[SparkException] {
       relation.parseRecords(testReader, oneRowRDD).collect()

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

@@ -16,10 +16,13 @@
 
 package za.co.absa.cobrix.spark.cobol.source.integration
 
+import org.apache.hadoop.fs.Path
 import org.apache.spark.SparkException
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.cobrix.cobol.reader.parameters.CobolParametersParser.{PARAM_ENABLE_INDEXES, PARAM_ENABLE_INDEX_CACHE}
 import za.co.absa.cobrix.spark.cobol.source.base.SparkTestBase
 import za.co.absa.cobrix.spark.cobol.source.fixtures.BinaryFileFixture
+import za.co.absa.cobrix.spark.cobol.source.index.IndexBuilder
 
 class Test37RecordLengthMappingSpec extends AnyWordSpec with SparkTestBase with BinaryFileFixture {
   private val copybook =
@@ -171,17 +174,15 @@ class Test37RecordLengthMappingSpec extends AnyWordSpec with SparkTestBase with
     }
 
     "work for data with offsets and indexes and index cache" in {
-      withTempBinFile("record_length_mapping", ".tmp", dataWithFileOffsets) { tempFile =>
+      withTempBinFile("record_length_mapping", ".tmp", dataSimple) { tempFile =>
         val expected = """{"SEG_ID":"A","TEXT":"123"},{"SEG_ID":"B","TEXT":"123456"},{"SEG_ID":"C","TEXT":"1234567"}"""
 
         val df = spark.read
           .format("cobol")
           .option("copybook_contents", copybook)
           .option("record_format", "F")
           .option("record_length_field", "SEG-ID")
-          .option("file_start_offset", 1)
-          .option("file_end_offset", 2)
-          .option("input_split_records", "2")
+          .option("input_split_records", "1")
           .option("enable_index_cache", "true")
           .option("pedantic", "true")
           .option("record_length_map", """{"A":4,"B":7,"C":8}""")
@@ -190,11 +191,35 @@ class Test37RecordLengthMappingSpec extends AnyWordSpec with SparkTestBase with
         val actualInitial = df.orderBy("SEG_ID").toJSON.collect().mkString(",")
         val actualCached = df.orderBy("SEG_ID").toJSON.collect().mkString(",")
 
+        val pathNameAsCached = s"file:$tempFile"
+
+        assert(IndexBuilder.indexCache.get(pathNameAsCached) != null)
+        assert(IndexBuilder.indexCache.get(pathNameAsCached).length == 2)
+
         assert(actualInitial == expected)
         assert(actualCached == expected)
       }
     }
 
+    "throw an exception when index caching is requested while indexes are turned off" in {
+      withTempBinFile("record_length_mapping", ".tmp", dataWithFileOffsets) { tempFile =>
+        val ex = intercept[IllegalArgumentException] {
+          spark.read
+            .format("cobol")
+            .option("copybook_contents", copybook)
+            .option("record_format", "F")
+            .option("record_length_field", "SEG-ID")
+            .option("enable_indexes", "false")
+            .option("enable_index_cache", "true")
+            .option("pedantic", "true")
+            .option("record_length_map", """{"A":4,"B":7,"C":8}""")
+            .load(tempFile)
+        }
+
+        assert(ex.getMessage == s"When '$PARAM_ENABLE_INDEXES' = false, '$PARAM_ENABLE_INDEX_CACHE' cannot be true.")
+      }
+    }
+
     "throw an exception for unknown mapping" in {
       withTempBinFile("record_length_mapping", ".tmp", dataSimple) { tempFile =>
         val df = spark.read