<commit_before>package jp.ne.opt.redshiftfake.read

import jp.ne.opt.redshiftfake._
import jp.ne.opt.redshiftfake.s3.{S3Location, S3Service}

/**
 * Provides features to download text from dataSource and read into rows.
 */
class Reader(copyCommand: CopyCommand, columnDefinitions: Seq[ColumnDefinition], s3Service: S3Service) {

  // TODO: Support more dataSources and formats
  def read(): Seq[Row] = {

    val contents = (copyCommand.copyFormat, copyCommand.dataSource) match {
      case (CopyFormat.Manifest(location), _) =>
        val rawManifest = s3Service.downloadAsString(location)(copyCommand.credentials)
        val manifest = new Manifest(rawManifest)
        manifest.files.map(s3Service.downloadAsString(_)(copyCommand.credentials))
      case (_, CopyDataSource.S3(location)) =>
        downloadAllAsStringFromS3(location)
    }

    copyCommand.copyFormat match {
      case CopyFormat.Json(Some(jsonpathsLocation)) =>
        val rawJsonpaths = s3Service.downloadAsString(S3Location(jsonpathsLocation.bucket, jsonpathsLocation.prefix))(copyCommand.credentials)
        val jsonpaths = new Jsonpaths(rawJsonpaths)

        (for {
          content <- contents
          line <- content.trim.lines
        } yield {
          val jsonReader = jsonpaths.mkReader(line.trim)
          val columns = columnDefinitions.zipWithIndex.map { case (_, n) =>
            Column(jsonReader.valueAt(n))
          }
          Row(columns)
        })(collection.breakOut)
      case CopyFormat.Manifest(manifestLocation) =>
        (for {
          content <- contents
          line <- content.trim.lines
        } yield {
          val csvReader = new CsvReader(line, copyCommand.delimiter, copyCommand.nullAs)
          csvReader.toRow
        })(collection.breakOut)
      case CopyFormat.Default =>
        (for {
          content <- contents
          line <- content.trim.lines
        } yield {
          val csvReader = new CsvReader(line, copyCommand.delimiter, copyCommand.nullAs)
          csvReader.toRow
        })(collection.breakOut)
      case _ => Nil
    }
  }

  private[this] def downloadAllAsStringFromS3(location: S3Location): Seq[String] = {
    val summaries = s3Service.lsRecurse(location)(copyCommand.credentials)
    summaries.map { summary =>
      s3Service.downloadAsString(S3Location(location.bucket, summary.getKey))(copyCommand.credentials)
    }
  }
}
<commit_msg>chore: CSV Reader case statement in on match clause for Manifest and Default<commit_after>package jp.ne.opt.redshiftfake.read

import jp.ne.opt.redshiftfake._
import jp.ne.opt.redshiftfake.s3.{S3Location, S3Service}

/**
 * Provides features to download text from dataSource and read into rows.
 */
class Reader(copyCommand: CopyCommand, columnDefinitions: Seq[ColumnDefinition], s3Service: S3Service) {

  // TODO: Support more dataSources and formats
  def read(): Seq[Row] = {

    val contents = (copyCommand.copyFormat, copyCommand.dataSource) match {
      case (CopyFormat.Manifest(location), _) =>
        val rawManifest = s3Service.downloadAsString(location)(copyCommand.credentials)
        val manifest = new Manifest(rawManifest)
        manifest.files.map(s3Service.downloadAsString(_)(copyCommand.credentials))
      case (_, CopyDataSource.S3(location)) =>
        downloadAllAsStringFromS3(location)
    }

    copyCommand.copyFormat match {
      case CopyFormat.Json(Some(jsonpathsLocation)) =>
        val rawJsonpaths = s3Service.downloadAsString(S3Location(jsonpathsLocation.bucket, jsonpathsLocation.prefix))(copyCommand.credentials)
        val jsonpaths = new Jsonpaths(rawJsonpaths)

        (for {
          content <- contents
          line <- content.trim.lines
        } yield {
          val jsonReader = jsonpaths.mkReader(line.trim)
          val columns = columnDefinitions.zipWithIndex.map { case (_, n) =>
            Column(jsonReader.valueAt(n))
          }
          Row(columns)
        })(collection.breakOut)
      case CopyFormat.Manifest(_) | CopyFormat.Default =>
        (for {
          content <- contents
          line <- content.trim.lines
        } yield {
          val csvReader = new CsvReader(line, copyCommand.delimiter, copyCommand.nullAs)
          csvReader.toRow
        })(collection.breakOut)
      case _ => Nil
    }
  }

  private[this] def downloadAllAsStringFromS3(location: S3Location): Seq[String] = {
    val summaries = s3Service.lsRecurse(location)(copyCommand.credentials)
    summaries.map { summary =>
      s3Service.downloadAsString(S3Location(location.bucket, summary.getKey))(copyCommand.credentials)
    }
  }
}

{'fraction_non_alphanumeric': 0.1019786910197869, 'fraction_numerical': 0.006958034355294629, 'mean_word_length': 3.0280210157618215, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6739975', 'n_tokens_mistral': 1339, 'n_tokens_neox': 1306, 'n_words': 251}