<commit_before>package com.bwsw.sj.module.input.csv

import java.io.ByteArrayOutputStream

import com.bwsw.sj.engine.core.entities.InputEnvelope
import com.bwsw.sj.engine.core.environment.InputEnvironmentManager
import com.bwsw.sj.engine.core.input.{InputStreamingExecutor, Interval}
import com.opencsv.CSVParserBuilder
import io.netty.buffer.ByteBuf
import org.apache.avro.SchemaBuilder
import org.apache.avro.generic.GenericData.Record
import org.apache.avro.generic.{GenericDatumWriter, GenericRecord}
import org.apache.avro.io.EncoderFactory

import scala.io.Source

/**
  * Executor for work with csv.
  *
  * @author <NAME>
  */
class CSVInputExecutor(manager: InputEnvironmentManager) extends InputStreamingExecutor[Array[Byte]](manager) {

  val outputStream: String = manager.options(CSVInputOptionNames.outputStream).asInstanceOf[String]
  val fallbackStream: String = manager.options(CSVInputOptionNames.fallbackStream).asInstanceOf[String]
  val lineSeparator: Byte = manager.options(CSVInputOptionNames.lineSeparator).asInstanceOf[String].head.toByte
  val fieldSeparator: Option[Char] = manager.options.get(CSVInputOptionNames.fieldSeparator).asInstanceOf[Option[String]].map(_.head)
  val quoteSymbol: Option[Char] = manager.options.get(CSVInputOptionNames.quoteSymbol).asInstanceOf[Option[String]].map(_.head)
  val encoding: String = manager.options(CSVInputOptionNames.encoding).asInstanceOf[String]
  val partition = 0

  val fields: Seq[String] = manager.options(CSVInputOptionNames.fields).asInstanceOf[Seq[String]]
  val fieldsNumber = fields.length
  val schema = {
    var scheme = SchemaBuilder.record("csv").fields()
    fields.foreach { field =>
      scheme = scheme.name(field).`type`().stringType().noDefault()
    }
    scheme.endRecord()
  }

  val writer = new GenericDatumWriter[GenericRecord](schema)
  val writerOutput = new ByteArrayOutputStream()
  val encoder = EncoderFactory.get().binaryEncoder(writerOutput, null)

  val uniqueKey = manager.options.get(CSVInputOptionNames.uniqueKey) match {
    case Some(uniqueFields: Seq[Any]) => uniqueFields.map(_.asInstanceOf[String])
    case _ => fields
  }

  val csvParser = {
    val csvParserBuilder = new CSVParserBuilder
    fieldSeparator.foreach(csvParserBuilder.withSeparator)
    quoteSymbol.foreach(csvParserBuilder.withQuoteChar)
    csvParserBuilder.build()
  }

  override def tokenize(buffer: ByteBuf): Option[Interval] = {
    val startIndex = buffer.readerIndex()
    val writerIndex = buffer.writerIndex()
    val endIndex = buffer.indexOf(startIndex, writerIndex, lineSeparator)

    if (endIndex != -1) Some(Interval(startIndex, endIndex))
    else None
  }

  override def parse(buffer: ByteBuf, interval: Interval): Option[InputEnvelope[Array[Byte]]] = {
    val length = interval.finalValue - interval.initialValue
    val dataBuffer = buffer.slice(interval.initialValue, length)
    val data = new Array[Byte](length)
    dataBuffer.getBytes(0, data)
    buffer.readerIndex(interval.finalValue + 1)
    val line = Source.fromBytes(data, encoding).mkString
    val values = csvParser.parseLine(line)

    if (values.length == fieldsNumber) {

      val record = new Record(schema)
      fields.zip(values).foreach { case (field, value) => record.put(field, value) }
      writer.write(record, encoder)
      encoder.flush()
      val serialized = writerOutput.toByteArray
      writerOutput.reset()
      val key = uniqueKey.foldLeft("") { (acc, field) => acc + "," + record.get(field) }

      Some(new InputEnvelope(
        s"$outputStream$key",
        Array((outputStream, partition)),
        true,
        serialized))
    } else {
      Some(new InputEnvelope(
        s"$fallbackStream,$line",
        Array((fallbackStream, partition)),
        true,
        data))
    }
  }
}<commit_msg>SJ-3382 change executor entity type from Array[Byte] to Record<commit_after>package com.bwsw.sj.module.input.csv

import com.bwsw.sj.engine.core.entities.InputEnvelope
import com.bwsw.sj.engine.core.environment.InputEnvironmentManager
import com.bwsw.sj.engine.core.input.{InputStreamingExecutor, Interval}
import com.opencsv.CSVParserBuilder
import io.netty.buffer.ByteBuf
import org.apache.avro.SchemaBuilder
import org.apache.avro.generic.GenericData.Record

import scala.io.Source

/**
  * Executor for work with csv.
  *
  * @author <NAME>
  */
class CSVInputExecutor(manager: InputEnvironmentManager) extends InputStreamingExecutor[Record](manager) {

  val outputStream: String = manager.options(CSVInputOptionNames.outputStream).asInstanceOf[String]
  val fallbackStream: String = manager.options(CSVInputOptionNames.fallbackStream).asInstanceOf[String]
  val lineSeparator: Byte = manager.options(CSVInputOptionNames.lineSeparator).asInstanceOf[String].head.toByte
  val fieldSeparator: Option[Char] = manager.options.get(CSVInputOptionNames.fieldSeparator).asInstanceOf[Option[String]].map(_.head)
  val quoteSymbol: Option[Char] = manager.options.get(CSVInputOptionNames.quoteSymbol).asInstanceOf[Option[String]].map(_.head)
  val encoding: String = manager.options(CSVInputOptionNames.encoding).asInstanceOf[String]
  val partition = 0

  val fields: Seq[String] = manager.options(CSVInputOptionNames.fields).asInstanceOf[Seq[String]]
  val fieldsNumber = fields.length
  val schema = {
    var scheme = SchemaBuilder.record("csv").fields()
    fields.foreach { field =>
      scheme = scheme.name(field).`type`().stringType().noDefault()
    }
    scheme.endRecord()
  }

  val fallbackFieldName = "data"
  val fallbackSchema = SchemaBuilder.record("fallback").fields()
    .name(fallbackFieldName).`type`().stringType().noDefault().endRecord()

  val uniqueKey = manager.options.get(CSVInputOptionNames.uniqueKey) match {
    case Some(uniqueFields: Seq[Any]) => uniqueFields.map(_.asInstanceOf[String])
    case _ => fields
  }

  val csvParser = {
    val csvParserBuilder = new CSVParserBuilder
    fieldSeparator.foreach(csvParserBuilder.withSeparator)
    quoteSymbol.foreach(csvParserBuilder.withQuoteChar)
    csvParserBuilder.build()
  }

  override def tokenize(buffer: ByteBuf): Option[Interval] = {
    val startIndex = buffer.readerIndex()
    val writerIndex = buffer.writerIndex()
    val endIndex = buffer.indexOf(startIndex, writerIndex, lineSeparator)

    if (endIndex != -1) Some(Interval(startIndex, endIndex))
    else None
  }

  override def parse(buffer: ByteBuf, interval: Interval): Option[InputEnvelope[Record]] = {
    val length = interval.finalValue - interval.initialValue
    val dataBuffer = buffer.slice(interval.initialValue, length)
    val data = new Array[Byte](length)
    dataBuffer.getBytes(0, data)
    buffer.readerIndex(interval.finalValue + 1)
    val line = Source.fromBytes(data, encoding).mkString
    val values = csvParser.parseLine(line)

    if (values.length == fieldsNumber) {
      val record = new Record(schema)
      fields.zip(values).foreach { case (field, value) => record.put(field, value) }
      val key = uniqueKey.foldLeft("") { (acc, field) => acc + "," + record.get(field) }

      Some(new InputEnvelope(
        s"$outputStream$key",
        Array((outputStream, partition)),
        true,
        record))
    } else {
      val record = new Record(schema)
      record.put(fallbackFieldName, line)
      Some(new InputEnvelope(
        s"$fallbackStream,$line",
        Array((fallbackStream, partition)),
        true,
        record))
    }
  }
}
{'fraction_non_alphanumeric': 0.11587578975668773, 'fraction_numerical': 0.0016131200430165346, 'mean_word_length': 4.446559297218156, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3528817', 'n_tokens_mistral': 2232, 'n_tokens_neox': 2203, 'n_words': 411}