<commit_before>/*
 * Copyright 2014–2016 SlamData Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package quasar.physical.sparkcore.fs

import quasar.Predef._
import quasar.fs._
import quasar.fs.QueryFile.ResultHandle
import quasar.fs.mount.FileSystemDef, FileSystemDef.DefErrT
import quasar.physical.sparkcore.fs.{readfile => corereadfile}
import quasar.physical.sparkcore.fs.{queryfile => corequeryfile}
import quasar.physical.sparkcore.fs.hdfs.writefile.HdfsWriteCursor
import quasar.effect._
import quasar.fs.ReadFile.ReadHandle
import quasar.fs.WriteFile.WriteHandle
import quasar.fp.TaskRef
import quasar.fp.free._
import quasar.fs.mount.FileSystemDef._
import quasar.EnvironmentError
import quasar.fs.mount.FileSystemDef._
import quasar.fs.mount.ConnectionUri
import quasar.contrib.pathy._

import org.apache.hadoop.fs.{FileSystem => HdfsFileSystem}
import org.apache.hadoop.conf.Configuration;
import java.net.URI

import pathy.Path._
import org.apache.spark._
import scalaz._, Scalaz._
import scalaz.concurrent.Task

package object hdfs {

  import corequeryfile.RddState

  val FsType = FileSystemType("spark-hdfs")

  type EffM1[A] = Coproduct[KeyValueStore[ResultHandle, RddState, ?], Read[SparkContext, ?], A]
  type Eff0[A] = Coproduct[KeyValueStore[ReadHandle, SparkCursor, ?], EffM1, A]
  type Eff1[A] = Coproduct[KeyValueStore[WriteHandle, HdfsWriteCursor, ?], Eff0, A]
  type Eff2[A] = Coproduct[Task, Eff1, A]
  type Eff3[A] = Coproduct[PhysErr, Eff2, A]
  type Eff[A]  = Coproduct[MonotonicSeq, Eff3, A]

  final case class SparkFSConf(sparkConf: SparkConf, hdfsUriStr: String, prefix: ADir)

  def parseUri(uri: ConnectionUri): DefinitionError \/ SparkFSConf = {

    def error(msg: String): DefinitionError \/ SparkFSConf =
      NonEmptyList(msg).left[EnvironmentError].left[SparkFSConf]

    def forge(master: String, hdfsUriStr: String, rootPath: String): DefinitionError \/ SparkFSConf =
      posixCodec.parseAbsDir(rootPath)
        .map { prefix =>
        SparkFSConf(new SparkConf().setMaster(master).setAppName("quasar"), hdfsUriStr, sandboxAbs(prefix))
      }.fold(error(s"Could not extrat a path from $rootPath"))(_.right[DefinitionError])

    uri.value.split('|').toList match {
      case master :: hdfsUriStr :: prefixPath :: Nil => forge(master, hdfsUriStr, prefixPath)
      case _ =>
        error("Missing master and prefixPath (seperated by |)" +
          " e.g spark://spark_host:port|hdfs://hdfs_host:port|/user/")
    }
  }

  final case class SparkHdfsFSDef[S[_]](run: Free[Eff, ?] ~> Free[S, ?], close: Free[S, Unit])

  private def fetchSparCoreJar: Task[String] = Task.delay {
    sys.env("QUASAR_HOME/sparkcore.jar")
  }

  private def sparkFsDef[S[_]](sparkConf: SparkConf)(implicit
    S0: Task :<: S,
    S1: PhysErr :<: S
  ): Free[S, SparkHdfsFSDef[S]] = {

    val genSc = fetchSparCoreJar.map { jar => 
      val sc = new SparkContext(sparkConf)
      sc.addJar(jar)
      sc
    }

    lift((TaskRef(0L) |@|
      TaskRef(Map.empty[ResultHandle, RddState]) |@|
      TaskRef(Map.empty[ReadHandle, SparkCursor]) |@|
      TaskRef(Map.empty[WriteHandle, HdfsWriteCursor]) |@|<commit_msg>Fix compilation issue<commit_after>/*
 * Copyright 2014–2016 SlamData Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package quasar.physical.sparkcore.fs

import quasar.Predef._
import quasar.fs._
import quasar.fs.QueryFile.ResultHandle
import quasar.fs.mount.FileSystemDef, FileSystemDef.DefErrT
import quasar.physical.sparkcore.fs.{readfile => corereadfile}
import quasar.physical.sparkcore.fs.{queryfile => corequeryfile}
import quasar.physical.sparkcore.fs.hdfs.writefile.HdfsWriteCursor
import quasar.effect._
import quasar.fs.ReadFile.ReadHandle
import quasar.fs.WriteFile.WriteHandle
import quasar.fp.TaskRef
import quasar.fp.free._
import quasar.fs.mount.FileSystemDef._
import quasar.EnvironmentError
import quasar.fs.mount.FileSystemDef._
import quasar.fs.mount.ConnectionUri
import quasar.contrib.pathy._

import java.net.URI
import scala.sys

import org.apache.hadoop.fs.{FileSystem => HdfsFileSystem}
import org.apache.hadoop.conf.Configuration
import pathy.Path._
import org.apache.spark._
import scalaz._, Scalaz._
import scalaz.concurrent.Task

package object hdfs {

  import corequeryfile.RddState

  val FsType = FileSystemType("spark-hdfs")

  type EffM1[A] = Coproduct[KeyValueStore[ResultHandle, RddState, ?], Read[SparkContext, ?], A]
  type Eff0[A] = Coproduct[KeyValueStore[ReadHandle, SparkCursor, ?], EffM1, A]
  type Eff1[A] = Coproduct[KeyValueStore[WriteHandle, HdfsWriteCursor, ?], Eff0, A]
  type Eff2[A] = Coproduct[Task, Eff1, A]
  type Eff3[A] = Coproduct[PhysErr, Eff2, A]
  type Eff[A]  = Coproduct[MonotonicSeq, Eff3, A]

  final case class SparkFSConf(sparkConf: SparkConf, hdfsUriStr: String, prefix: ADir)

  def parseUri(uri: ConnectionUri): DefinitionError \/ SparkFSConf = {

    def error(msg: String): DefinitionError \/ SparkFSConf =
      NonEmptyList(msg).left[EnvironmentError].left[SparkFSConf]

    def forge(master: String, hdfsUriStr: String, rootPath: String): DefinitionError \/ SparkFSConf =
      posixCodec.parseAbsDir(rootPath)
        .map { prefix =>
        SparkFSConf(new SparkConf().setMaster(master).setAppName("quasar"), hdfsUriStr, sandboxAbs(prefix))
      }.fold(error(s"Could not extrat a path from $rootPath"))(_.right[DefinitionError])

    uri.value.split('|').toList match {
      case master :: hdfsUriStr :: prefixPath :: Nil => forge(master, hdfsUriStr, prefixPath)
      case _ =>
        error("Missing master and prefixPath (seperated by |)" +
          " e.g spark://spark_host:port|hdfs://hdfs_host:port|/user/")
    }
  }

  final case class SparkHdfsFSDef[S[_]](run: Free[Eff, ?] ~> Free[S, ?], close: Free[S, Unit])

  private def fetchSparCoreJar: Task[String] = Task.delay {
    sys.env("QUASAR_HOME") + "/sparkcore.jar"
  }

  private def sparkFsDef[S[_]](sparkConf: SparkConf)(implicit
    S0: Task :<: S,
    S1: PhysErr :<: S
  ): Free[S, SparkHdfsFSDef[S]] = {

    val genSc = fetchSparCoreJar.map { jar => 
      val sc = new SparkContext(sparkConf)
      sc.addJar(jar)
      sc
    }

    lift((TaskRef(0L) |@|
      TaskRef(Map.empty[ResultHandle, RddState]) |@|
      TaskRef(Map.empty[ReadHandle, SparkCursor]) |@|
      TaskRef(Map.empty[WriteHandle, HdfsWriteCursor]) |@|
{'fraction_non_alphanumeric': 0.11968482543132726, 'fraction_numerical': 0.006792555359326179, 'mean_word_length': 4.311688311688312, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '140880', 'n_tokens_mistral': 2564, 'n_tokens_neox': 2426, 'n_words': 575}