#805 Fix PR suggestions.

Author: yruslan

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

@@ -950,6 +950,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.isIndexGenerationNeeded)(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

@@ -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

@@ -18,6 +18,7 @@ package za.co.absa.cobrix.spark.cobol.source.integration
 
 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
 
@@ -195,6 +196,27 @@ class Test37RecordLengthMappingSpec extends AnyWordSpec with SparkTestBase with
       }
     }
 
+    "throw an exception when index caching is requested while indexes are turned off" in {
+      withTempBinFile("record_length_mapping", ".tmp", dataWithFileOffsets) { tempFile =>
+        val expected = """{"SEG_ID":"A","TEXT":"123"},{"SEG_ID":"B","TEXT":"123456"},{"SEG_ID":"C","TEXT":"1234567"}"""
+
+        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