fix json array source file + init file source specifications

fupelaqu · fupelaqu · commit 2d707a475481 · 2025-12-05T14:57:34.000+01:00
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/file/package.scala b/core/src/main/scala/app/softnetwork/elastic/client/file/package.scala
@@ -3,7 +3,7 @@ package app.softnetwork.elastic.client
 import akka.NotUsed
 import akka.stream.scaladsl.Source
 import com.fasterxml.jackson.annotation.JsonInclude
-import com.fasterxml.jackson.core.{JsonParser, JsonToken}
+import com.fasterxml.jackson.core.{JsonFactory, JsonParser, JsonToken}
 import com.fasterxml.jackson.databind.node.ObjectNode
 import com.fasterxml.jackson.databind.{
   DeserializationFeature,
@@ -15,13 +15,14 @@ import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
 import org.apache.avro.generic.GenericRecord
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hadoop.fs.{FSDataInputStream, FileSystem, Path}
 import org.apache.parquet.avro.AvroParquetReader
 import org.apache.parquet.hadoop.ParquetReader
 import org.apache.parquet.hadoop.util.HadoopInputFile
-import io.delta.standalone.{DeltaLog, Snapshot}
+import io.delta.standalone.DeltaLog
 import io.delta.standalone.data.{CloseableIterator, RowRecord}
 import io.delta.standalone.types._
+import org.apache.parquet.io.SeekableInputStream
 import org.slf4j.{Logger, LoggerFactory}
 
 import java.io.{BufferedReader, InputStream, InputStreamReader}
@@ -391,8 +392,8 @@ package object file {
 
     override def format: FileFormat = JsonArray
 
-    /** Reads a JSON Array file and streams each element
-      */
+    private val jsonFactory = new JsonFactory()
+
     override def fromFile(
       filePath: String,
       bufferSize: Int = 500
@@ -401,29 +402,34 @@ package object file {
       conf: Configuration = hadoopConfiguration
     ): Source[String, NotUsed] = {
 
-      // Validate file before processing
       validateFile(filePath)
 
       var elementCount = 0L
       val startTime = System.currentTimeMillis()
 
       Source
-        .unfoldResource[String, (BufferedReader, JsonParser)](
-          // Create: Open the file and JSON parser
+        .unfoldResource[String, (InputStream, JsonParser)](
+          // Create: Open file via Hadoop and create JSON parser
           create = () => {
-            logger.info(s"📂 Opening JSON Array file: $filePath")
+            logger.info(s"📂 Opening JSON Array file via Hadoop: $filePath")
             Try {
-              val is: InputStream = HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
-              val reader = new BufferedReader(new InputStreamReader(is, "UTF-8"))
-              val factory = mapper.getFactory
-              val parser = factory.createParser(reader)
-
-              // Vérifier que c'est bien un array
-              if (parser.nextToken() != JsonToken.START_ARRAY) {
-                throw new IllegalArgumentException(s"File is not a JSON array: $filePath")
+              val is: SeekableInputStream =
+                HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+
+              // Create Jackson parser on top of Hadoop SeekableInputStream
+              val parser = jsonFactory.createParser(is)
+
+              // Expect array start
+              val token = parser.nextToken()
+              if (token != JsonToken.START_ARRAY) {
+                is.close()
+                throw new IllegalArgumentException(
+                  s"Expected JSON array, but found: ${token}. File: $filePath"
+                )
               }
 
-              (reader, parser)
+              logger.info(s"📊 Started parsing JSON Array via Hadoop FS")
+              (is, parser)
             } match {
               case Success(result) => result
               case Failure(ex) =>
@@ -432,48 +438,63 @@ package object file {
             }
           },
 
-          // Read: Read the next JSON object from the array
+          // Read: Parse next element from array
           read = { case (_, parser) =>
             blocking {
               Try {
-                if (parser.nextToken() == JsonToken.START_OBJECT) {
+                val token = parser.nextToken()
+
+                if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
                   elementCount += 1
+
                   if (elementCount % 10000 == 0) {
                     val elapsed = (System.currentTimeMillis() - startTime) / 1000.0
                     val throughput = elementCount / elapsed
-                    logger.info(
-                      f"📊 Read $elementCount elements from JSON Array ($throughput%.2f elements/sec)"
-                    )
+                    logger.info(f"📊 Parsed $elementCount elements ($throughput%.2f elements/sec)")
                   }
 
-                  // Read the full object
-                  val node = mapper.readTree(parser)
+                  // Parse current element as JsonNode
+                  val node: JsonNode = mapper.readTree(parser)
                   Some(mapper.writeValueAsString(node))
-                } else {
+
+                } else if (token == JsonToken.END_ARRAY || token == null) {
                   val elapsed = (System.currentTimeMillis() - startTime) / 1000.0
                   logger.info(
                     s"✅ Finished reading JSON Array: $elementCount elements in ${elapsed}s"
                   )
                   None
+
+                } else {
+                  // Skip unexpected tokens
+                  logger.warn(s"⚠️  Unexpected token in JSON Array: $token")
+                  None
                 }
               } match {
                 case Success(result) => result
                 case Failure(ex) =>
-                  logger.error(s"❌ Error reading JSON Array element at position $elementCount", ex)
+                  logger.error(s"❌ Error parsing JSON Array element at position $elementCount", ex)
                   None
               }
             }
           },
 
-          // Close: Close parser and reader
-          close = { case (reader, parser) =>
+          // Close: Close parser and Hadoop input stream
+          close = { case (inputStream, parser) =>
             Try {
-              parser.close()
-              reader.close()
+              parser.close() // This also closes the underlying stream
             } match {
-              case Success(_) => logger.debug(s"🔒 Closed JSON Array reader for: $filePath")
+              case Success(_) =>
+                logger.debug(s"🔒 Closed JSON Array parser for: $filePath")
+              case Failure(ex) =>
+                logger.warn(s"⚠️  Failed to close JSON Array parser: ${ex.getMessage}")
+            }
+
+            // Ensure Hadoop stream is closed
+            Try(inputStream.close()) match {
+              case Success(_) =>
+                logger.debug(s"🔒 Closed Hadoop input stream for: $filePath")
               case Failure(ex) =>
-                logger.warn(s"⚠️  Failed to close JSON Array reader: ${ex.getMessage}")
+                logger.warn(s"⚠️  Failed to close Hadoop input stream: ${ex.getMessage}")
             }
           }
         )
@@ -504,7 +525,6 @@ package object file {
                 throw new IllegalArgumentException(s"File is not a JSON array: $filePath")
               }
 
-              import scala.jdk.CollectionConverters._
               arrayNode.elements().asScala.map(node => mapper.writeValueAsString(node)).toList
             } finally {
               is.close()
@@ -849,18 +869,17 @@ package object file {
       val startTime = System.currentTimeMillis()
 
       Source
-        .unfoldResource[String, (DeltaLog, CloseableIterator[RowRecord], Snapshot)](
+        .unfoldResource[String, CloseableIterator[RowRecord]](
           create = () => {
             logger.info(s"📂 Opening Delta Lake table at version $version: $filePath")
             val deltaLog = DeltaLog.forTable(conf, filePath)
             val snapshot = deltaLog.getSnapshotForVersionAsOf(version)
 
             logger.info(s"📊 Delta table version $version, files: ${snapshot.getAllFiles.size()}")
 
-            val iterator = snapshot.open()
-            (deltaLog, iterator, snapshot)
+            snapshot.open()
           },
-          read = { case (deltaLog, iterator, snapshot) =>
+          read = iterator =>
             blocking {
               Try {
                 if (iterator.hasNext) {
@@ -886,11 +905,8 @@ package object file {
                   logger.error(s"❌ Error reading Delta row at position $rowCount", ex)
                   None
               }
-            }
-          },
-          close = { case (deltaLog, iterator, snapshot) =>
-            Try(iterator.close())
-          }
+            },
+          close = iterator => Try(iterator.close())
         )
         .buffer(bufferSize, akka.stream.OverflowStrategy.backpressure)
     }
@@ -961,7 +977,7 @@ package object file {
       } else if (lowerPath.endsWith(".jsonl") || lowerPath.endsWith(".ndjson")) {
         Json
       } else if (lowerPath.endsWith(".json")) {
-        // Distinguer JSON Lines vs JSON Array en lisant le premier caractère
+        // Distinguishing JSON Lines vs JSON Array by reading the first character
         detectJsonType(filePath)
       } else if (isDeltaTable(filePath)) {
         Delta
@@ -978,7 +994,7 @@ package object file {
       Try {
         val fs = FileSystem.get(conf)
         val deltaLogPath = new Path(filePath, "_delta_log")
-        fs.exists(deltaLogPath) && fs.isDirectory(deltaLogPath)
+        fs.exists(deltaLogPath) && fs.getFileStatus(deltaLogPath).isDirectory
       }.getOrElse(false)
     }
 
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/file/FileSourceSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/file/FileSourceSpec.scala
@@ -0,0 +1,136 @@
+package app.softnetwork.elastic.client.file
+
+import akka.actor.ActorSystem
+import akka.stream.scaladsl.Sink
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.time.{Seconds, Span}
+
+import scala.concurrent.ExecutionContext
+
+class FileSourceSpec extends AnyWordSpec with Matchers with ScalaFutures with BeforeAndAfterAll {
+
+  implicit val system: ActorSystem = ActorSystem("FileSourceTest")
+  implicit val ec: ExecutionContext = system.dispatcher
+  implicit val patience: PatienceConfig = PatienceConfig(timeout = Span(10, Seconds))
+
+  "FileFormatDetector" should {
+    "detect Parquet files" in {
+      FileFormatDetector.detect("data/file.parquet") shouldBe Parquet
+      FileFormatDetector.detect("data/file.parq") shouldBe Parquet
+    }
+
+    "detect JSON files" in {
+      FileFormatDetector.detect("data/file.jsonl") shouldBe Json
+      FileFormatDetector.detect("data/file.ndjson") shouldBe Json
+    }
+
+    "detect unknown files" in {
+      FileFormatDetector.detect("data/file.txt") shouldBe Unknown
+    }
+
+    "throw on unknown files with detectOrThrow" in {
+      assertThrows[IllegalArgumentException] {
+        FileFormatDetector.detectOrThrow("data/file.txt")
+      }
+    }
+  }
+
+  "JsonFileSource" should {
+    "read JSON lines file" in {
+      // Create a temporary file for testing
+      val tempFile = java.io.File.createTempFile("test", ".jsonl")
+      tempFile.deleteOnExit()
+
+      val writer = new java.io.PrintWriter(tempFile)
+      writer.println("""{"id":1,"name":"Alice"}""")
+      writer.println("""{"id":2,"name":"Bob"}""")
+      writer.close()
+
+      val result = JsonFileSource
+        .fromFile(tempFile.getAbsolutePath)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result should have size 2
+      result.head should include("Alice")
+      result.last should include("Bob")
+    }
+  }
+
+  "JsonFileFactory" should {
+    "read JSON lines file" in {
+      // Create a temporary file for testing
+      val tempFile = java.io.File.createTempFile("test", ".jsonl")
+      tempFile.deleteOnExit()
+
+      val writer = new java.io.PrintWriter(tempFile)
+      writer.println("""{"id":1,"name":"Alice","nested":{"age":30}}""")
+      writer.println("""{"id":2,"name":"Bob","nested":{"age":25}}""")
+      writer.close()
+
+      val result = FileSourceFactory
+        .fromFile(tempFile.getAbsolutePath)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result should have size 2
+      result.head should include("Alice")
+      result.last should include("Bob")
+    }
+
+    "read JSON Array file (single line)" in {
+      val tempFile = java.io.File.createTempFile("test", ".json")
+      tempFile.deleteOnExit()
+
+      val writer = new java.io.PrintWriter(tempFile)
+      writer.println(
+        """[{"id":1,"name":"Alice","nested": {"age":30}},{"id":2,"name":"Bob","nested":{"age":25}}]"""
+      )
+      writer.close()
+
+      val result = FileSourceFactory
+        .fromFile(tempFile.getAbsolutePath, format = JsonArray)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result should have size 2
+      result.head should include("Alice")
+      result.last should include("Bob")
+    }
+
+    "read JSON Array file (multi-line)" in {
+      val tempFile = java.io.File.createTempFile("test", ".json")
+      tempFile.deleteOnExit()
+
+      val writer = new java.io.PrintWriter(tempFile)
+      writer.println("[")
+      writer.println("""  {
+          "id":1,
+          "name":"Alice",
+          "nested":{
+            "age":30
+           }
+         },""")
+      writer.println("""  {"id":2,"name":"Bob","nested":{"age":25}}""")
+      writer.println("]")
+      writer.close()
+
+      val result = FileSourceFactory
+        .fromFile(tempFile.getAbsolutePath)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result should have size 2
+      result.head should include("Alice")
+      result.last should include("Bob")
+    }
+  }
+
+  override def afterAll(): Unit = {
+    system.terminate()
+    super.afterAll()
+  }
+}
