#251 Fix glob support and divibility check for large amount of files.

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

@@ -96,11 +96,9 @@ private [source] object CobolScanners {
 
   private def areThereNonDivisibleFiles(sourceDir: String, hadoopConfiguration: Configuration, divisor: Int): Boolean = {
 
-    val THRESHOLD_DIR_LENGTH_FOR_SINGLE_FILE_CHECK = 50
-
     val fileSystem = FileSystem.get(hadoopConfiguration)
 
-    if (FileUtils.getNumberOfFilesInDir(sourceDir, fileSystem) < THRESHOLD_DIR_LENGTH_FOR_SINGLE_FILE_CHECK) {
+    if (FileUtils.getNumberOfFilesInDir(sourceDir, fileSystem) < FileUtils.THRESHOLD_DIR_LENGTH_FOR_SINGLE_FILE_CHECK) {
       FileUtils.findAndLogAllNonDivisibleFiles(sourceDir, divisor, fileSystem) > 0
     }
     else {

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

@@ -37,6 +37,8 @@ object FileUtils {
 
   private val logger = LoggerFactory.getLogger(this.getClass)
 
+  val THRESHOLD_DIR_LENGTH_FOR_SINGLE_FILE_CHECK = 50
+
   private val hiddenFileFilter = new PathFilter() {
     def accept(p: Path): Boolean = {
       val name = p.getName
@@ -195,9 +197,10 @@ object FileUtils {
     */
   def findAndLogFirstNonDivisibleFile(sourceDir: String, divisor: Long, fileSystem: FileSystem): Boolean = {
 
-    val allFiles = fileSystem.listStatus(new Path(sourceDir))
+    val allFiles = expandDirectories(fileSystem, fileSystem.globStatus(new Path(sourceDir), hiddenFileFilter))
 
-    val firstNonDivisibleFile = allFiles.find(isNonDivisible(_, divisor))
+    val firstNonDivisibleFile = allFiles.take(THRESHOLD_DIR_LENGTH_FOR_SINGLE_FILE_CHECK)
+      .find(isNonDivisible(_, divisor))
 
     if (firstNonDivisibleFile.isDefined) {
       logger.error(s"File ${firstNonDivisibleFile.get.getPath} IS NOT divisible by $divisor.")

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

@@ -18,6 +18,9 @@ package za.co.absa.cobrix.spark.cobol.source.fixtures
 
 import java.io.{DataOutputStream, File, FileOutputStream}
 import java.nio.charset.Charset
+import java.nio.file.{Files, Path}
+
+import org.apache.commons.io.{FileSystemUtils, FileUtils}
 
 /**
   * This fixture adds ability for a unit test to create temporary files for using them in the tests.
@@ -67,5 +70,14 @@ trait BinaryFileFixture {
     tempFile.delete
   }
 
+  def withTempDirectory(prefix: String)(f: String => Unit): Unit = {
+    val tmpPath = Files.createTempDirectory(prefix)
+    val pathStr = tmpPath.toAbsolutePath.toString
+
+    f(pathStr)
+
+    FileUtils.deleteDirectory(new File(pathStr))
+  }
+
 }
 

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.cobrix.spark.cobol.source.regression
+
+import java.io.{DataOutputStream, File, FileOutputStream}
+import java.nio.file.{FileSystem, Files, Path, Paths}
+
+import org.apache.hadoop.fs.{FileSystem => HadoopFs}
+import org.scalatest.FunSuite
+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.utils.FileUtils
+
+class Test07IgnoreHiddenFiles extends FunSuite with BinaryFileFixture with SparkTestBase {
+  private val fileSystem = HadoopFs.get(spark.sparkContext.hadoopConfiguration)
+
+  test("Test findAndLogFirstNonDivisibleFile() finds a file") {
+    withTempDirectory("testHidden1") { tmpDir =>
+      createFileSize1(Files.createFile(Paths.get(tmpDir, "a")))
+      assert(FileUtils.findAndLogFirstNonDivisibleFile(tmpDir, 2, fileSystem))
+      assert(FileUtils.findAndLogAllNonDivisibleFiles(tmpDir, 2, fileSystem) == 1)
+    }
+  }
+
+  test("Test findAndLogFirstNonDivisibleFile() ignores a hidden file") {
+    withTempDirectory("testHidden1") { tmpDir =>
+      createFileSize1(Files.createFile(Paths.get(tmpDir, ".a")))
+      assert(!FileUtils.findAndLogFirstNonDivisibleFile(tmpDir, 2, fileSystem))
+      assert(FileUtils.findAndLogAllNonDivisibleFiles(tmpDir, 2, fileSystem) == 0)
+    }
+  }
+
+  test("Test findAndLogFirstNonDivisibleFile() ignores a  hidden file in a nested dir") {
+    withTempDirectory("testHidden3") { tmpDir =>
+      Files.createDirectory(Paths.get(tmpDir, "dir1"))
+      createFileSize1(Files.createFile(Paths.get(tmpDir, "dir1", ".b2")))
+      assert(!FileUtils.findAndLogFirstNonDivisibleFile(tmpDir, 2, fileSystem))
+      assert(FileUtils.findAndLogAllNonDivisibleFiles(tmpDir, 2, fileSystem) == 0)
+    }
+  }
+
+  test("Test findAndLogFirstNonDivisibleFile() ignores a hidden dir") {
+    withTempDirectory("testHidden4") { tmpDir =>
+      Files.createDirectory(Paths.get(tmpDir, ".dir2"))
+      createFileSize1(Files.createFile(Paths.get(tmpDir, ".dir2", "c1")))
+      assert(!FileUtils.findAndLogFirstNonDivisibleFile(tmpDir, 2, fileSystem))
+      assert(FileUtils.findAndLogAllNonDivisibleFiles(tmpDir, 2, fileSystem) == 0)
+    }
+  }
+
+  test("Test findAndLogFirstNonDivisibleFile() works with globbing") {
+    withTempDirectory("testHidden1") { tmpDir =>
+      createFileSize1(Files.createFile(Paths.get(tmpDir, "a")))
+      assert(FileUtils.findAndLogFirstNonDivisibleFile(s"$tmpDir/*", 2, fileSystem))
+      assert(FileUtils.findAndLogAllNonDivisibleFiles(tmpDir, 2, fileSystem) == 1)
+    }
+  }
+
+
+  private def createFileSize1(path: Path): Unit = {
+    val file = new File(path.toAbsolutePath.toString)
+    val ostream = new DataOutputStream(new FileOutputStream(file))
+    val content = Array[Byte](0.toByte)
+    ostream.write(content)
+    ostream.close()
+  }
+}