<commit_before>package io.eels.component.orc

import com.sksamuel.exts.Logging
import io.eels.schema.StructType
import io.eels.{Row, Sink, SinkWriter}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import com.sksamuel.exts.OptionImplicits._

case class OrcSink(path: Path,
                   overwrite: Boolean = false,
                   bloomFilterColumns: Seq[String] = Nil,
                   rowIndexStride: Option[Int] = None)
                  (implicit fs: FileSystem, conf: Configuration) extends Sink with Logging {

  def withBloomFilterColumns(bloomFilterColumns: Seq[String]): OrcSink = copy(bloomFilterColumns = bloomFilterColumns)
  def withRowIndexStride(stride: Int): OrcSink = copy(rowIndexStride = stride.some)
  def withOverwrite(overwrite: Boolean): OrcSink = copy(overwrite = overwrite)

  override def writer(schema: StructType): SinkWriter = new SinkWriter {

    if (overwrite && fs.exists(path))
      fs.delete(path, false)

    val writer = new OrcWriter(path, schema, bloomFilterColumns, rowIndexStride)

    override def write(row: Row): Unit = writer.write(row)
    override def close(): Unit = writer.close()
  }
}<commit_msg>Added permissions to orc sink #226<commit_after>package io.eels.component.orc

import com.sksamuel.exts.Logging
import com.sksamuel.exts.OptionImplicits._
import io.eels.schema.StructType
import io.eels.{Row, Sink, SinkWriter}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.permission.FsPermission
import org.apache.hadoop.fs.{FileSystem, Path}

case class OrcSink(path: Path,
                   overwrite: Boolean = false,
                   bloomFilterColumns: Seq[String] = Nil,
                   permission: Option[FsPermission] = None,
                   inheritPermissions: Option[Boolean] = None,
                   rowIndexStride: Option[Int] = None)
                  (implicit fs: FileSystem, conf: Configuration) extends Sink with Logging {

  def withBloomFilterColumns(bloomFilterColumns: Seq[String]): OrcSink = copy(bloomFilterColumns = bloomFilterColumns)
  def withRowIndexStride(stride: Int): OrcSink = copy(rowIndexStride = stride.some)
  def withOverwrite(overwrite: Boolean): OrcSink = copy(overwrite = overwrite)
  def withPermission(permission: FsPermission): OrcSink = copy(permission = Option(permission))
  def withInheritPermission(inheritPermissions: Boolean): OrcSink = copy(inheritPermissions = Option(inheritPermissions))

  override def writer(schema: StructType): SinkWriter = new SinkWriter {

    if (overwrite && fs.exists(path))
      fs.delete(path, false)

    val writer = new OrcWriter(path, schema, bloomFilterColumns, rowIndexStride)

    override def write(row: Row): Unit = writer.write(row)
    
    override def close(): Unit = {
      writer.close()
      permission match {
        case Some(perm) => fs.setPermission(path, perm)
        case None =>
          if (inheritPermissions.getOrElse(false)) {
            val permission = fs.getFileStatus(path.getParent).getPermission
            fs.setPermission(path, permission)
          }
      }
    }
  }
}
{'fraction_non_alphanumeric': 0.09920255183413078, 'fraction_numerical': 0.0009569377990430622, 'mean_word_length': 3.558139534883721, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '4937342', 'n_tokens_mistral': 937, 'n_tokens_neox': 917, 'n_words': 214}