What is the best way to handle parsing of large files ?

[mackoj]

Hi,

What do you think is the best way to load large files ? swift-parsing doesn't seems to recommend / document a preferred way to handle parsing with large files.

• URLSession.shared.bytes(for: <#T##URLRequest#>) / URLSession.shared.bytes(from: <#T##URL#>)
• let handle = FileHandle(forReadingAtPath: <#T##String#>) and defer { handle?.close() }
• String(contentsOf: <#T##URL#>)[...].utf8

Maybe I misunderstood this but I think that the library pushes us to make a parser that works on the whole file at once but I don't think it's the right approach for large files. The issue is that for a parser to work in a continuous consuming mode it has to be thought out this way.

Should swift-parsing offer some convenience for this?

Thanks,

FWIW I did build something to reduce this issue in my project.

It's a protocol that represents the ProgressiveParsing and something that implements the file consumption.

public protocol ProgressiveParsing {
  associatedtype PartialType : Codable
  associatedtype CompleteType : Codable

  static func parser(inputFileURL: Foundation.URL, cacheKey: String?, loadFromCache: Bool, urlSession: URLSession) -> AsyncThrowingStream<Progression<CompleteType, PartialType>, Error>
}

public enum Progression<CompleteType: Codable, PartialType: Codable> {
  case start
  case parsingProgress(current: Int, total: Int, partial: PartialType)
  case fullFile(CompleteType, fromCache: Bool)
}

The FileConsumer protocol presents a unified way to handle this.

public protocol ProgressiveParsing {
  associatedtype PartialType : Codable
  associatedtype CompleteType : Codable

  static func parser(inputFileURL: Foundation.URL, cacheKey: String?, loadFromCache: Bool, urlSession: URLSession) -> AsyncThrowingStream<Progression<CompleteType, PartialType>, Error>
}

public enum Progression<CompleteType: Codable, PartialType: Codable> {
  case start
  case parsingProgress(current: Int, total: Int, partial: PartialType)
  case fullFile(CompleteType, fromCache: Bool)
}

FileConsumer can:

• download a file and store it in URLCache
• parse the download file and store it in NSCache
• load parsed file from NSCache
• load url ressource from URLCache

import Foundation

public actor FileConsumer {
  public enum ProgressState<Partial: Codable, Complete: Codable> {
    case start
    case progress(current: Int, total: Int, lineNbr: Int, lineLenght: Int, generated: [Partial])
    case finished(Complete, fromCache: Bool)
  }
    
  public func saveInCache<Complete: Codable>(url: URL, cacheKey: String? = nil, _ toCache: Complete) async throws
  
  public func downloadThenParse<Complete: Codable>(
    url: URL,
    cacheOutput: Complete.Type,
    cacheKey: String? = nil,
    loadFromCache: Bool = true,
    urlSession: URLSession = .shared,
    parse: @escaping (inout Substring.UTF8View) throws -> Complete
  ) -> AsyncThrowingStream<ProgressState<Complete, Complete>, Error>
  
  public func downloadAndParse<Partial: Codable, Complete: Codable>(
    url: URL,
    cacheOutput: Complete.Type,
    cacheKey: String? = nil,
    loadFromCache: Bool = true,
    useFullLines: Bool = true,
    lineIteration: Int = 1,
    urlSession: URLSession = .shared,
    parse: @escaping (inout Substring.UTF8View) throws -> [Partial]
  ) -> AsyncThrowingStream<ProgressState<Partial, Complete>, Error>

}

[stephencelis]

@mackoj We'd love to enhance parsing to work well with async sequences, but haven't figured out an approach that quite generically fits into the existing library. If you explore any promising approaches we'd love to hear about them!

[mackoj]

I have tried without finding something that fits into the existing library for the moment I will continue to think about it.

let parseProgress = someParser.stream(
inputFileURL: inputFileURL, 
transform: Transform(apply: { $0[...].utf8 }, unapply: { String($0)! }), 
parse: someParser.parse
)

Working code the but the API is meh

import Foundation
import Parsing

/*
 class ParserBox<T> {
     var content: T
     init(content: T) {
         self.content = T
     }
 }
 
 let cache = NSCache<NSString, ParserBox<Output>>()

 // get from cache
 guard let cacheKey = cacheKey  else { continuation.finish() }
                                                          
 if loadFromCache, let object = cache.object(forKey: NSString(string: cacheKey)) {
   continuation.yield(.fullFile(object.content, fromCache: true))
   continuation.finish()
   return
 }

 // set to cache
cache.setValue(data, forKey: "toto")
 */

 
 
enum Download {
  case thenParse
  case andParse
}

struct Transform<Input> {
  let apply: (_ input: String) -> Input
  let unapply: (_ input: Input) -> String
  
  init(apply: @escaping (_ input: String) -> Input, unapply: @escaping (_ input: Input) -> String) {
      self.apply = apply
      self.unapply = unapply
  }
}

extension Parser {
  func stream(inputFileURL url: Foundation.URL, transform: Transform<Input>, parse: @escaping (_ input: inout Input) throws -> Output, download: Download = .andParse, urlSession: URLSession = .shared, useFullLines: Bool = true, lineIteration: Int = 1) -> AsyncThrowingStream<ParserProgression<Output>, Error> where Output: Codable  {
    return AsyncThrowingStream { continuation in
      continuation.yield(.start)

      switch download {
      case .thenParse:
        var filePath: URL = url
        if url.isFileURL == false {
          let request = URLRequest(url: url)
          let (fpath, _) = try await urlSession.download(for: request)
          filePath = fpath
        }
        let fileContent = try String(contentsOf: filePath) // this is at least sub-optimal
        var tmp = transform.apply(fileContent)
        let parsed = try parse(&tmp)
        continuation.yield(.finished(parsed))
        continuation.finish()

      case .andParse:
        var current: Int = 0
        let request = URLRequest(url: url)
        let (bytes, response) = try await urlSession.bytes(for: request)
        let total = Int(response.expectedContentLength)
        var accumulateur: String = ""
        var output: [Output] = []
        for try await line in bytes.computedLines(useFullLines, lineIteration) {
          current += Int(line.lengthOfBytes(using: .utf8))
          accumulateur.append(line)
          // essaye de parser
          var subString = transform.apply(accumulateur)
          // Cannot convert value of type 'Substring.UTF8View' to expected argument type 'Self.Input'
          let parsedLine: Output = try parse(&subString)
          continuation.yield(.parsingProgress(current: current, total: total, partial: parsedLine))
          output.append(parsedLine)
          accumulateur = transform.unapply(subString)
        }
        continuation.finish()
      }
    }
  }
}

public enum ParserProgression<CompleteType: Codable> {
  case start
  case parsingProgress(current: Int, total: Int, partial: CompleteType)
  case finished(CompleteType)
}

extension URLSession.AsyncBytes {
  
  /// This works like lines but with CR and and empty lines in the buffer
  /// When the file contain a line like "Hello World\nI'm Alive"
  /// Instead of sending "Hello World" like .line api  it will send "Hello World\n"
  func computedLines(_ useFullline: Bool = false, _ nbr: Int = 1) -> AsyncThrowingStream<String, Error> {
    return AsyncThrowingStream { continuation in
      var acc: String = ""
      var nbLine = 0
      for try await char in self.characters {
        if char == "\n" {
          nbLine += 1
        }
        if char == "\n", nbLine == nbr {
          continuation.yield(acc + (useFullline == true ? "\n" : ""))
          acc = ""
          nbLine = 0
        } else {
          acc.append(char)
        }
      }
      if acc.isEmpty == false {
        continuation.yield(acc)
      }
      continuation.finish()
    }
  }
}

[mackoj]

Without the parse argument it's already better...

    let streamParser/* : AsyncThrowingStream<ParserProgression<Output>, Error> */ = someParser.stream(
      inputFileURL: inputFileURL,
      transform: Transform(apply: { $0[...].utf8 }, unapply: { String($0)! })
    )

[mackoj]

I still don't like the transform(that serves to convert the file into a format that is supported by the parser) we need this because we are consuming the file and keeping a buffer of data that feeds the parser continuously.

enum Download {
  case thenParse
  case andParse
}

struct Transform<Input> {
  let apply: (_ input: String) -> Input
  let unapply: (_ input: Input) -> String
  
  init(apply: @escaping (_ input: String) -> Input, unapply: @escaping (_ input: Input) -> String) {
      self.apply = apply
      self.unapply = unapply
  }
}

extension Parser {
  func stream(inputFileURL url: Foundation.URL, transform: Transform<Input>, download: Download = .andParse, urlSession: URLSession = .shared, useFullLines: Bool = true, lineIteration: Int = 1) -> AsyncThrowingStream<ParserProgression<Output>, Error> where Output: Codable  {
    return AsyncThrowingStream { continuation in
      continuation.yield(.start)

      switch download {
      case .thenParse:
        var filePath: URL = url
        if url.isFileURL == false {
          let request = URLRequest(url: url)
          let (fpath, _) = try await urlSession.download(for: request)
          filePath = fpath
        }
        let fileContent = try String(contentsOf: filePath) // this is at least sub-optimal
        var tmp = transform.apply(fileContent)
        let parsed = try parse(&tmp)
        continuation.yield(.finished(parsed))
        continuation.finish()

      case .andParse:
        var current: Int = 0
        let request = URLRequest(url: url)
        let (bytes, response) = try await urlSession.bytes(for: request)
        let total = Int(response.expectedContentLength)
        var accumulateur: String = ""
        for try await line in bytes.computedLines(useFullLines, lineIteration) {
          current += Int(line.lengthOfBytes(using: .utf8))
          accumulateur.append(line)
          // essaye de parser
          var subString = transform.apply(accumulateur)
          // Cannot convert value of type 'Substring.UTF8View' to expected argument type 'Self.Input'
          let parsedLine: Output = try parse(&subString)
          continuation.yield(.parsingProgress(current: current, total: total, partial: parsedLine))
          accumulateur = transform.unapply(subString)
        }
        continuation.finish()
      }
    }
  }
}

public enum ParserProgression<CompleteType: Codable> {
  case start
  case parsingProgress(current: Int, total: Int, partial: CompleteType)
  case finished(CompleteType)
}

extension URLSession.AsyncBytes {
  
  /// This works like lines but with CR and and empty lines in the buffer
  /// When the file contain a line like "Hello World\nI'm Alive"
  /// Instead of sending "Hello World" like .line api  it will send "Hello World\n"
  func computedLines(_ useFullline: Bool = false, _ nbr: Int = 1) -> AsyncThrowingStream<String, Error> {
    return AsyncThrowingStream { continuation in
      var acc: String = ""
      var nbLine = 0
      for try await char in self.characters {
        if char == "\n" {
          nbLine += 1
        }
        if char == "\n", nbLine == nbr {
          continuation.yield(acc + (useFullline == true ? "\n" : ""))
          acc = ""
          nbLine = 0
        } else {
          acc.append(char)
        }
      }
      if acc.isEmpty == false {
        continuation.yield(acc)
      }
      continuation.finish()
    }
  }

}

[mackoj]

Hi @stephencelis,

I did arrive at an API that I find simple and that integrates pretty well with the actual API of swift-parsing.
Instead of calling .parse on a parser we can call .steam it return an AsyncThrowingStream<ParserStreamProgression<Output>, Error>.

/* AsyncThrowingStream<ParserStreamProgression<Output>, Error> */
for try await parsed in someParser.stream(inputFileURL: inputFileURL) {
  switch parsed {
   case .start:
     print("The parsing processing process has started")
   case let .progress(current, total, parsed):
     print("Have found \(parsed.count) elements.")
   case let .finish(parsed):
     print("Have found \(parsed.count) elements.")
  }
}

The steam function is configurable it expects:

• at least the URL of the resource that it will download and parse simultaneously
• a transform function that will make the download data into a type compatible with the parser(if the parser expect Substring.UTF8View as input a default transform exist)
• the method of download and processing can either download and parse simultaneously or download the whole file and then parse the whole file at once
• the URLSession can be provided if the user wants to use URLCaching for example

If you chose to use the download and parse simultaneously mode you can ask for full lines(aka lines with the \n at the end) and the number of lines you want to send to the parser.

Below there is an example implementation of it, it is not perfect but it provides an idea of the feasibility.

Rest of the code

import Foundation
import Parsing

extension Parser {
  public func stream(
    inputFileURL url: Foundation.URL,
    transform: Transform<Input>,
    download: SteamDownloadProcess = .downloadAndParse(useFullLines: true, lineIteration: 1),
    urlSession: URLSession = .shared
  ) -> AsyncThrowingStream<ParserStreamProgression<Output>, Error> where Output: Codable  {
    return AsyncThrowingStream { continuation in
      continuation.yield(.start)
      
      switch download {
      case .downloadThenParse:
        var filePath: URL = url
        if url.isFileURL == false {
          let request = URLRequest(url: url)
          let (fpath, _) = try await urlSession.download(for: request)
          filePath = fpath
        }
        let fileContent = try String(contentsOf: filePath) // this need to be improved
        var tmp = transform.apply(fileContent)
        let parsed = try parse(&tmp)
        continuation.yield(.finish(parsed))
        
      case let .downloadAndParse(useFullLines, lineIteration):
        var current: Int = 0
        let request = URLRequest(url: url)
        let (bytes, response) = try await urlSession.bytes(for: request)
        let total = Int(response.expectedContentLength)
        var accumulateur: String = ""
        for try await line in bytes.computedLines(useFullLines, lineIteration) {
          current += Int(line.lengthOfBytes(using: .utf8))
          accumulateur.append(line)
          var subString = transform.apply(accumulateur)
          let parsedLine: Output = try parse(&subString)
          continuation.yield(.progress(current: current, total: total, parsed: parsedLine))
          accumulateur = transform.unapply(subString)
        }
      }
      
      continuation.finish()
    }
  }
}

extension Parser where Input == Substring.UTF8View {
  public func stream(
    inputFileURL url: Foundation.URL,
    download: SteamDownloadProcess = .downloadAndParse(useFullLines: true, lineIteration: 1),
    urlSession: URLSession = .shared
  ) -> AsyncThrowingStream<ParserStreamProgression<Output>, Error> where Output: Codable  {
    return stream(inputFileURL: url, transform: .utf8, download: download, urlSession: urlSession)
  }
}

public enum SteamDownloadProcess {
  case downloadThenParse
  case downloadAndParse(useFullLines: Bool/*true: means don't consume new lines*/, lineIteration: Int/* number of lines send to the parser */)
}

public enum ParserStreamProgression<Output: Codable> {
  case start
  case progress(current: Int, total: Int, parsed: Output)
  case finish(Output)
}

import Foundation

extension URLSession.AsyncBytes {
  
  /// This works like lines but with CR and and empty lines in the buffer
  /// When the file contain a line like "Hello World\n\nI'm Alive"
  /// Instead of sending "Hello World" like .line api  it will send "Hello World\n" then "\n" then "I'm Alive"
  public func computedLines(_ useFullline: Bool = false, _ nbr: Int = 1) -> AsyncThrowingStream<String, Error> {
    return AsyncThrowingStream { continuation in
      var acc: String = ""
      var nbLine = 0
      for try await char in self.characters {
        if char == "\n" {
          nbLine += 1
        }
        if char == "\n", nbLine == nbr {
          continuation.yield(acc + (useFullline == true ? "\n" : ""))
          acc = ""
          nbLine = 0
        } else {
          acc.append(char)
        }
      }
      if acc.isEmpty == false {
        continuation.yield(acc)
      }
      continuation.finish()
    }
  }

}

import Foundation

public struct Transform<Input> {
  let apply: (_ input: String) -> Input
  let unapply: (_ input: Input) -> String
  
  init(apply: @escaping (_ input: String) -> Input, unapply: @escaping (_ input: Input) -> String) {
    self.apply = apply
    self.unapply = unapply
  }
}

extension Transform where Input == Substring.UTF8View {
  public static var utf8 = Transform(apply: { $0[...].utf8 }, unapply: { String($0)! })
}

extension AsyncThrowingStream where Failure == Error {
  public init(_ build: @escaping (Continuation) async throws -> Void) {
    self.init { continuation in
      Task {
        do {
          try await build(continuation)
          continuation.finish()
        } catch {
          continuation.finish(throwing: error)
        }
      }
    }
  }
}

[mackoj]

It still miss support for:

• URLRequest
• URLCache example
• ParsedData cache example