<commit_before>package org.sparkpipe.rdd

import java.io.IOException
import java.nio.charset.CodingErrorAction
import scala.collection.mutable.ArrayBuffer
import scala.io.{Source, Codec}
import scala.reflect.ClassTag
import scala.sys.process.{Process, ProcessBuilder, ProcessIO}
import scala.util.control.NonFatal
import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.rdd.RDD

/**
 * Piped RDD with encoding.
 * Encoding can be specified as string or Codec.
 * `strict` parameter allows to silence errors during reading. They will be returned as a part of
 * dataset with prefix "[ERROR]". Otherwise exception is thrown, for invalid command, as well as
 * malicious input.
 */
private[rdd] class EncodePipedRDD[T: ClassTag](
    prev: RDD[T],
    private val encoding: String,
    private val strict: Boolean
) extends RDD[String](prev) {

    /**
     * Allows to specify codec instead of encoding string.
     * although `on...` actions are dropped, specify them using `strict` parameter
     */
    def this(prev: RDD[T], codec: Codec, strict: Boolean) = this(prev, codec.toString, strict)

    override def getPartitions: Array[Partition] = firstParent[T].partitions

    override def compute(split: Partition, context: TaskContext): Iterator[String] = {
        // set warning if `file.encoding` and `sun.jnu.encoding` properties do not match specified
        // encoding, also display how to resolve problem.
        val fileEncodingProp = Option(System.getProperty("file.encoding"))
        val sunJnuEncodingProp = Option(System.getProperty("sun.jnu.encoding"))
        if (!fileEncodingProp.isDefined || !sunJnuEncodingProp.isDefined ||
            !fileEncodingProp.get.equals(encoding) || !sunJnuEncodingProp.get.equals(encoding)) {
            logWarning("System encoding is different from " + encoding + ". Even encoding will " +
                "be applied, driver program may display different result. It is recommended to " +
                "set configuration options 'spark.driver.extraJavaOptions' and/or " +
                "'spark.executor.extraJavaOptions' to be '-Dfile.encoding=YOUR_VALUE " +
                "-Dsun.jnu.encoding=YOUR_VALUE'"
            )
        }

        // rule to apply on incorrect command
        // if true, it will enforce exception, otherwise, it will silence it returning as output
        val executionRule = strict
        // rule to apply on malformed exception
        val malformedRule = if (strict) CodingErrorAction.REPORT else CodingErrorAction.REPLACE

        // create buffers for output and errors, those are always string buffers
        val logbuf: ArrayBuffer[String] = ArrayBuffer()
        val errbuf: ArrayBuffer[String] = ArrayBuffer()
        // codec (applies only for output) and error codec
        val codec: Codec = Codec(encoding).onMalformedInput(malformedRule)
        val errCodec: Codec = Codec.UTF8.onMalformedInput(CodingErrorAction.REPLACE)

        for (elem <- firstParent[T].iterator(split, context)) {
            val logger: ProcessIO = new ProcessIO(
                (input => input.close()),
                (output => {
                    // output can fail with malformed exception or any arbitrary exception
                    // need to catch it and push to the error buffer
                    // use `toString` to fetch type of exception
                    try {
                        Source.fromInputStream(output)(codec).getLines().foreach(line =>
                            logbuf.append(line)
                        )
                    } catch {
                        case NonFatal(e) => errbuf.append("[ERROR] " + e.toString())
                    } finally {
                        output.close()
                    }
                }),
                (err => {
                    // we apply UTF-8 encoding and REPLACE malformed rule for error, as it is a
                    // general case, and we have to return error without generating another
                    // exception.
                    try {
                        Source.fromInputStream(err)(errCodec).getLines().foreach(errline =>
                            errbuf.append("[ERROR] " + errline)
                        )
                    } finally {
                        err.close()
                    }
                }),
                daemonizeThreads=false
            )
            // tokenize complex command, use resulting buffer as iterator
            val cmds = EncodePipedRDD.tokenize(elem.toString)
            // create buffered process from all the commands
            var bufferedProcess: ProcessBuilder = null
            cmds.foreach(cmd => {
                require(cmd.nonEmpty, "Tokenizer returned empty command from " + elem.toString)
                if (bufferedProcess == null) {
                    bufferedProcess = Process(cmd)
                } else {
                    bufferedProcess = bufferedProcess.#|(Process(cmd))
                }
            })
            // execute process, similar to Java's `process.waitFor()`
            val exit = bufferedProcess.run(logger).exitValue()

            // check for errors and strict rule, if something has happened, return first error
            if (executionRule && errbuf.nonEmpty) {
                throw new IOException("Subprocess exited with status " + exit.toString + ". " +
                    "Failed for elem: " + elem.toString + ", reason: " + errbuf.head)
            }
        }
        // merge two buffers into one. Strategy: if error buffer is empty, return log buffer, else
        // merge them, even if log buffer is empty.
        val merged = if (errbuf.isEmpty) logbuf else logbuf ++ errbuf
        merged.toIterator
    }
}<commit_msg>closes #12, error log added<commit_after>package org.sparkpipe.rdd

import java.io.IOException
import java.nio.charset.CodingErrorAction
import scala.collection.mutable.ArrayBuffer
import scala.io.{Source, Codec}
import scala.reflect.ClassTag
import scala.sys.process.{Process, ProcessBuilder, ProcessIO}
import scala.util.control.NonFatal
import org.apache.spark.{Partition, TaskContext}
import org.apache.spark.rdd.RDD

/**
 * Piped RDD with encoding.
 * Encoding can be specified as string or Codec.
 * `strict` parameter allows to silence errors during reading. They will be returned as a part of
 * dataset with prefix "[ERROR]". Otherwise exception is thrown, for invalid command, as well as
 * malicious input.
 */
private[rdd] class EncodePipedRDD[T: ClassTag](
    prev: RDD[T],
    private val encoding: String,
    private val strict: Boolean
) extends RDD[String](prev) {

    /**
     * Allows to specify codec instead of encoding string.
     * although `on...` actions are dropped, specify them using `strict` parameter
     */
    def this(prev: RDD[T], codec: Codec, strict: Boolean) = this(prev, codec.toString, strict)

    override def getPartitions: Array[Partition] = firstParent[T].partitions

    override def compute(split: Partition, context: TaskContext): Iterator[String] = {
        // set warning if `file.encoding` and `sun.jnu.encoding` properties do not match specified
        // encoding, also display how to resolve problem.
        val fileEncodingProp = Option(System.getProperty("file.encoding"))
        val sunJnuEncodingProp = Option(System.getProperty("sun.jnu.encoding"))
        if (!fileEncodingProp.isDefined || !sunJnuEncodingProp.isDefined ||
            !fileEncodingProp.get.equals(encoding) || !sunJnuEncodingProp.get.equals(encoding)) {
            logWarning("System encoding is different from " + encoding + ". Even encoding will " +
                "be applied, driver program may display different result. It is recommended to " +
                "set configuration options 'spark.driver.extraJavaOptions' and/or " +
                "'spark.executor.extraJavaOptions' to be '-Dfile.encoding=YOUR_VALUE " +
                "-Dsun.jnu.encoding=YOUR_VALUE'"
            )
        }

        // rule to apply on incorrect command
        // if true, it will enforce exception, otherwise, it will silence it returning as output
        val executionRule = strict
        // rule to apply on malformed exception
        val malformedRule = if (strict) CodingErrorAction.REPORT else CodingErrorAction.REPLACE

        // create buffers for output and errors, those are always string buffers
        val logbuf: ArrayBuffer[String] = ArrayBuffer()
        val errbuf: ArrayBuffer[String] = ArrayBuffer()
        // codec (applies only for output) and error codec
        val codec: Codec = Codec(encoding).onMalformedInput(malformedRule)
        val errCodec: Codec = Codec.UTF8.onMalformedInput(CodingErrorAction.REPLACE)

        for (elem <- firstParent[T].iterator(split, context)) {
            val logger: ProcessIO = new ProcessIO(
                (input => input.close()),
                (output => {
                    // output can fail with malformed exception or any arbitrary exception
                    // need to catch it and push to the error buffer
                    // use `toString` to fetch type of exception
                    try {
                        Source.fromInputStream(output)(codec).getLines().foreach(line =>
                            logbuf.append(line)
                        )
                    } catch {
                        case NonFatal(e) => errbuf.append("[ERROR] " + e.toString())
                    } finally {
                        output.close()
                    }
                }),
                (err => {
                    // we apply UTF-8 encoding and REPLACE malformed rule for error, as it is a
                    // general case, and we have to return error without generating another
                    // exception.
                    try {
                        Source.fromInputStream(err)(errCodec).getLines().foreach(errline =>
                            errbuf.append("[ERROR] " + errline)
                        )
                    } finally {
                        err.close()
                    }
                }),
                daemonizeThreads=false
            )
            // tokenize complex command, use resulting buffer as iterator
            val cmds = EncodePipedRDD.tokenize(elem.toString)
            // create buffered process from all the commands
            var bufferedProcess: ProcessBuilder = null
            cmds.foreach(cmd => {
                require(cmd.nonEmpty, "Tokenizer returned empty command from " + elem.toString)
                if (bufferedProcess == null) {
                    bufferedProcess = Process(cmd)
                } else {
                    bufferedProcess = bufferedProcess.#|(Process(cmd))
                }
            })
            // execute process, similar to Java's `process.waitFor()`
            val exit = bufferedProcess.run(logger).exitValue()

            // check for errors and strict rule, if something has happened, return first error
            // if rule is strict then throw exception, otherwise log error message
            if (errbuf.nonEmpty) {
                val errorMsg = "Subprocess exited with status " + exit.toString + ". " +
                    "Failed for elem: " + elem.toString + ", reason: " + errbuf.head
                if (executionRule) {
                    throw new IOException(errorMsg)
                } else {
                    logError(errorMsg)
                }
            }
        }
        // merge two buffers into one. Strategy: if error buffer is empty, return log buffer, else
        // merge them, even if log buffer is empty.
        val merged = if (errbuf.isEmpty) logbuf else logbuf ++ errbuf
        merged.toIterator
    }
}
{'fraction_non_alphanumeric': 0.08864313058273075, 'fraction_numerical': 0.0005104210974053594, 'mean_word_length': 1.929479192623972, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, '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': '3397305', 'n_tokens_mistral': 2965, 'n_tokens_neox': 2819, 'n_words': 913}